From 5814c20dd995fd65e56cc3ad9a99b518b17af2fe Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Sun, 18 Oct 2026 20:16:18 +0800
Subject: [PATCH 1/7] feat(taskplugin): persist catalog and recover pinned host
 state

---
 docs/task-plugin-host.md            |  20 +-
 main.go                             |   3 +
 model/main.go                       |   2 +
 model/task_plugin_catalog.go        | 232 ++++++++++++++++++
 model/task_plugin_catalog_test.go   | 303 ++++++++++++++++++++++++
 pkg/taskplugin/catalog.go           | 353 ++++++++++++++++++++++++++++
 pkg/taskplugin/catalog_test.go      | 214 +++++++++++++++++
 pkg/taskplugin/host.go              |  19 +-
 pkg/taskplugin/package.go           |   6 +-
 service/task_plugin_catalog.go      |  52 ++++
 service/task_plugin_catalog_test.go |  71 ++++++
 11 files changed, 1267 insertions(+), 8 deletions(-)
 create mode 100644 model/task_plugin_catalog.go
 create mode 100644 model/task_plugin_catalog_test.go
 create mode 100644 pkg/taskplugin/catalog.go
 create mode 100644 pkg/taskplugin/catalog_test.go
 create mode 100644 service/task_plugin_catalog.go
 create mode 100644 service/task_plugin_catalog_test.go

diff --git a/docs/task-plugin-host.md b/docs/task-plugin-host.md
index f2af0290..83229136 100644
--- a/docs/task-plugin-host.md
+++ b/docs/task-plugin-host.md
@@ -11,11 +11,26 @@
 - 路由只匹配 manifest 声明的上游模型；省略 `route.Models` 表示继承 manifest 的范围，指定时进一步收窄。缺失或未知模型拒绝绑定。
 - `BindMapped` 固定插件版本、包摘要、原始及映射后模型，以及宿主拥有的 usage contract；`Resolve` 使用历史包恢复任务，不自动切换最新版本。
 - `AttachTaskPluginBinding` 校验用户、Token、渠道、请求、模型及冻结用量契约，写入 `Task.PrivateData`。调用方必须在发起付费请求前通过现有事务保存任务；该函数自身不执行数据库写入或上游提交。
-- 重启时必须重新加载所需历史包，并在开放流量前恢复宿主的撤销记录。缺失历史包、无效契约、身份变化或已撤销版本均拒绝恢复，不自动重提任务。当前没有自动保存插件目录或撤销清单的管理服务。
+- 持久目录在开放流量前恢复历史包和撤销记录。缺失历史包、摘要不匹配、无效契约、身份变化或已撤销版本均拒绝恢复，不自动重提任务。
 
 The loader accepts at most 1 MiB and verifies an externally trusted SHA-256 of the whole document. Batch installation publishes packages and routes together; a conflict leaves the previous generation intact. A key/version cannot silently change contents. Routes match only manifest-declared upstream models; an omitted `route.Models` inherits that set, while an explicit list narrows it. Missing or unknown models cannot bind. Bindings freeze package, usage contract, client model alias and upstream model. Recovery resolves historical packages rather than the latest version.
 
-The service validates task ownership and the frozen usage contract before attaching private metadata. The caller must persist it in the existing transaction before any billable request. The host must reload retained historical packages and revocations before serving traffic after restart. Missing or revoked identities fail closed and never cause automatic submission. Automatic catalog/revocation storage is not implemented here.
+The service validates task ownership and the frozen usage contract before attaching private metadata. The caller must persist it in the existing transaction before any billable request. The durable catalog restores historical packages and revocations before startup serves traffic. Missing, altered or revoked identities fail closed and never cause automatic submission.
+
+## 持久目录与恢复 / Durable catalog and recovery
+
+`taskplugin.OpenCatalog(ctx, model.NewTaskPluginCatalogStore(db))` 加载内部目录，`catalog.Host()` 提供既有绑定与恢复能力。应用启动由 `InitializeTaskPluginCatalog` 完成恢复；缺表、损坏包或无效状态会阻止启动。目录仍没有公开管理路由或自动加载 JS 的行为。
+
+- `Install` 批量校验并保存完整包，原始文档字节和摘要保持不变；同 key/version 不允许替换内容。安装版本成为该 key 的启用版本，历史版本保留。
+- `SetEnabled(false)` 停止新任务绑定，旧任务仍能按历史版本恢复。`Activate` 显式切换已有版本，不自动选择“最新”版本。
+- `Revoke` 永久撤销指定版本，新绑定和现有 session 的下一次使用均被拒绝；接口不提供撤销回滚或历史版本删除。
+- `task_plugin_catalog_heads` 的版本号条件更新与 `task_plugin_versions` 的变更在同一事务内提交。事务成功后一次发布宿主状态；存储失败、结果未知或刷新失败会暂停宿主，成功 `Refresh` 后恢复。
+- 包文档在 MySQL 使用 LONGTEXT，在 SQLite/PostgreSQL 使用 TEXT；版本身份使用固定长度摘要索引。升级会新增这两张表，备份与恢复应包含两表的一致状态。
+- 其他节点的变更通过显式 `Refresh` 读取。本阶段没有后台同步任务，不保证跨节点实时撤销；后续执行器接入时必须定义刷新时机。目录容量沿用 64 MiB 历史包原始字节上限。
+
+`OpenCatalog` restores the internal catalog through a GORM store. Startup restores it before serving traffic and stops on missing tables, corrupt bytes or invalid state. Batch installation preserves exact package bytes and immutable key/version identities. Disabling blocks new admission but preserves historical task recovery; activation explicitly selects a retained version. Revocation is permanent and invalidates existing sessions on their next use. Historical versions are never automatically deleted or replaced.
+
+A singleton revision CAS serializes metadata transactions across nodes, including concurrent bootstrap. Publication follows commit; uncertain writes or failed refreshes suspend the managed host until a successful refresh. MySQL stores documents as LONGTEXT; SQLite and PostgreSQL use TEXT. Migrations add `task_plugin_catalog_heads` and `task_plugin_versions`; back up and restore both consistently. Other nodes' changes require explicit `Refresh`; background synchronization and immediate multi-node revocation are outside this stage. Retained document bytes remain capped at 64 MiB. The catalog registers no public routes and never evaluates JavaScript.
 
 ## HTTP 能力 / HTTP capability
 
@@ -37,6 +52,7 @@ Limits are four calls, 15 seconds per call, a 30-second scope deadline, 1 MiB re
 go test ./pkg/taskplugin ./pkg/taskusage -count=1
 go test ./service -run TestTaskPluginHost -count=1
 go test ./model -run TestTaskPluginHost -count=1
+go test ./pkg/taskplugin ./model ./service -run 'TestCatalog|TestTaskPluginCatalog' -count=1
 ```
 
 这些合成测试不代表真实供应商、JS 隔离、多节点撤销传播或 MySQL/PostgreSQL 实库验收已通过。
diff --git a/main.go b/main.go
index 6dfae480..7ad55e52 100644
--- a/main.go
+++ b/main.go
@@ -447,6 +447,9 @@ func InitResources() error {
 	}
 
 	model.CheckSetup()
+	if err := service.InitializeTaskPluginCatalog(context.Background(), model.DB); err != nil {
+		return fmt.Errorf("initialize task plugin catalog: %w", err)
+	}
 
 	// Initialize options, should after model.InitDB()
 	model.InitOptionMap()
diff --git a/model/main.go b/model/main.go
index e60cf140..01e47acc 100644
--- a/model/main.go
+++ b/model/main.go
@@ -421,6 +421,8 @@ func migrateDB() error {
 		&QuotaDataSnapshot{},
 		&QuotaDataSnapshotRetry{},
 		&Task{},
+		&TaskPluginCatalogHead{},
+		&TaskPluginVersion{},
 		&Model{},
 		&Vendor{},
 		&PrefillGroup{},
diff --git a/model/task_plugin_catalog.go b/model/task_plugin_catalog.go
new file mode 100644
index 00000000..7a029b7a
--- /dev/null
+++ b/model/task_plugin_catalog.go
@@ -0,0 +1,232 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package model
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"math"
+
+	"github.com/MAX-API-Next/MAX-API/common"
+	"github.com/MAX-API-Next/MAX-API/pkg/taskplugin"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"gorm.io/gorm/logger"
+	"gorm.io/gorm/schema"
+)
+
+// One revision serializes catalog mutations across nodes, including bootstrap.
+// MutationID distinguishes a duplicate INSERT under MySQL CLIENT_FOUND_ROWS.
+type TaskPluginCatalogHead struct {
+	ID         uint   `gorm:"primaryKey;autoIncrement:false"`
+	Revision   int64  `gorm:"not null"`
+	MutationID string `gorm:"size:64;not null"`
+}
+
+type taskPluginDocument string
+
+func (taskPluginDocument) GormDataType() string { return "string" }
+func (taskPluginDocument) GormDBDataType(db *gorm.DB, _ *schema.Field) string {
+	if db.Dialector.Name() == "mysql" {
+		return "longtext"
+	}
+	return "text"
+}
+
+// Identity hashes key/version to keep the primary index within MySQL 5.7's
+// older utf8mb4 limits. The pinned document remains the identity authority.
+// Documents are immutable, private, and never changed by metadata updates.
+type TaskPluginVersion struct {
+	Identity string             `gorm:"size:64;primaryKey" json:"-"`
+	Digest   string             `gorm:"size:64;not null" json:"-"`
+	Document taskPluginDocument `gorm:"not null" json:"-"`
+	Active   bool               `gorm:"not null"`
+	Enabled  bool               `gorm:"not null"`
+	Revoked  bool               `gorm:"not null"`
+}
+
+type TaskPluginCatalogStore struct{ db *gorm.DB }
+
+func NewTaskPluginCatalogStore(db *gorm.DB) *TaskPluginCatalogStore {
+	if db == nil {
+		return &TaskPluginCatalogStore{}
+	}
+	// A driver error can embed SQL arguments containing a package's source.
+	return &TaskPluginCatalogStore{db: db.Session(&gorm.Session{Logger: db.Logger.LogMode(logger.Silent)})}
+}
+
+var _ taskplugin.CatalogStore = (*TaskPluginCatalogStore)(nil)
+
+func loadTaskPluginVersions(tx *gorm.DB) ([]taskplugin.StoredPackage, error) {
+	var records []TaskPluginVersion
+	if err := tx.Order("identity").Find(&records).Error; err != nil {
+		return nil, err
+	}
+	result := make([]taskplugin.StoredPackage, 0, len(records))
+	for _, record := range records {
+		entry := taskplugin.StoredPackage{Document: []byte(record.Document), Digest: record.Digest, Active: record.Active, Enabled: record.Enabled, Revoked: record.Revoked}
+		identity, err := taskPluginStorageIdentity(entry)
+		if err != nil || identity != record.Identity {
+			return nil, taskplugin.ErrCatalogInvalid
+		}
+		result = append(result, entry)
+	}
+	return result, nil
+}
+
+func taskPluginStorageIdentity(entry taskplugin.StoredPackage) (string, error) {
+	p, err := taskplugin.LoadPackage(bytes.NewReader(entry.Document), entry.Digest)
+	if err != nil {
+		return "", taskplugin.ErrCatalogInvalid
+	}
+	manifest := p.Manifest()
+	digest := sha256.Sum256([]byte(manifest.Key + "@" + manifest.Version))
+	return hex.EncodeToString(digest[:]), nil
+}
+
+func (store *TaskPluginCatalogStore) Load(ctx context.Context) (taskplugin.CatalogSnapshot, error) {
+	if store == nil || store.db == nil || ctx == nil {
+		return taskplugin.CatalogSnapshot{}, taskplugin.ErrCatalogUnavailable
+	}
+	var snapshot taskplugin.CatalogSnapshot
+	err := store.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var head TaskPluginCatalogHead
+		err := tx.First(&head, 1).Error
+		missing := errors.Is(err, gorm.ErrRecordNotFound)
+		if err != nil && !missing {
+			return err
+		}
+		if !missing && (head.Revision <= 0 || head.MutationID == "") {
+			return taskplugin.ErrCatalogInvalid
+		}
+		rows, err := loadTaskPluginVersions(tx)
+		if err != nil {
+			return err
+		}
+		if missing && len(rows) != 0 {
+			return taskplugin.ErrCatalogInvalid
+		}
+		var after TaskPluginCatalogHead
+		err = tx.First(&after, 1).Error
+		// Under READ COMMITTED, reject a mixed snapshot. MySQL's consistent
+		// snapshot and SQLite's transaction also satisfy this invariant.
+		if missing {
+			if !errors.Is(err, gorm.ErrRecordNotFound) {
+				return taskplugin.ErrCatalogConflict
+			}
+		} else if err != nil || head != after {
+			return taskplugin.ErrCatalogConflict
+		}
+		snapshot = taskplugin.CatalogSnapshot{Revision: head.Revision, Packages: rows}
+		return nil
+	})
+	if err != nil {
+		if errors.Is(err, taskplugin.ErrCatalogInvalid) || errors.Is(err, taskplugin.ErrCatalogConflict) {
+			return taskplugin.CatalogSnapshot{}, err
+		}
+		return taskplugin.CatalogSnapshot{}, taskplugin.ErrCatalogUnavailable
+	}
+	return snapshot, nil
+}
+
+func (store *TaskPluginCatalogStore) CompareAndSwap(ctx context.Context, revision int64, packages []taskplugin.StoredPackage) (int64, error) {
+	if store == nil || store.db == nil || ctx == nil {
+		return 0, taskplugin.ErrCatalogUnavailable
+	}
+	if revision < 0 || revision == math.MaxInt64 {
+		return 0, taskplugin.ErrCatalogInvalid
+	}
+	next := taskplugin.CatalogSnapshot{Revision: revision + 1, Packages: packages}
+	if err := taskplugin.ValidateCatalogTransition(taskplugin.CatalogSnapshot{}, next); err != nil {
+		return 0, err
+	}
+	mutationID := common.GetUUID()
+	err := store.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if revision == 0 {
+			head := TaskPluginCatalogHead{ID: 1, Revision: 1, MutationID: mutationID}
+			if err := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&head).Error; err != nil {
+				return err
+			}
+			var claimed TaskPluginCatalogHead
+			if err := tx.First(&claimed, 1).Error; err != nil {
+				return err
+			}
+			if claimed.MutationID != mutationID || claimed.Revision != 1 {
+				return taskplugin.ErrCatalogConflict
+			}
+		} else {
+			result := tx.Model(&TaskPluginCatalogHead{}).Where("id = ? AND revision = ?", 1, revision).
+				Updates(map[string]any{"revision": revision + 1, "mutation_id": mutationID})
+			if result.Error != nil {
+				return result.Error
+			}
+			if result.RowsAffected != 1 {
+				return taskplugin.ErrCatalogConflict
+			}
+		}
+		previousRows, err := loadTaskPluginVersions(tx)
+		if err != nil {
+			return err
+		}
+		if revision == 0 && len(previousRows) != 0 {
+			return taskplugin.ErrCatalogInvalid
+		}
+		previous := taskplugin.CatalogSnapshot{Revision: revision, Packages: previousRows}
+		if err := taskplugin.ValidateCatalogTransition(previous, next); err != nil {
+			return err
+		}
+		known := make(map[string]bool, len(previousRows))
+		for _, old := range previousRows {
+			identity, err := taskPluginStorageIdentity(old)
+			if err != nil {
+				return err
+			}
+			known[identity] = true
+		}
+		for _, entry := range next.Packages {
+			identity, err := taskPluginStorageIdentity(entry)
+			if err != nil {
+				return err
+			}
+			if known[identity] {
+				if err := tx.Model(&TaskPluginVersion{}).Where("identity = ?", identity).
+					Updates(map[string]any{"active": entry.Active, "enabled": entry.Enabled, "revoked": entry.Revoked}).Error; err != nil {
+					return err
+				}
+			} else {
+				record := TaskPluginVersion{Identity: identity, Digest: entry.Digest, Document: taskPluginDocument(entry.Document), Active: entry.Active, Enabled: entry.Enabled, Revoked: entry.Revoked}
+				if err := tx.Create(&record).Error; err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		if errors.Is(err, taskplugin.ErrCatalogConflict) || errors.Is(err, taskplugin.ErrCatalogInvalid) {
+			return 0, err
+		}
+		return 0, taskplugin.ErrCatalogUnavailable
+	}
+	return revision + 1, nil
+}
diff --git a/model/task_plugin_catalog_test.go b/model/task_plugin_catalog_test.go
new file mode 100644
index 00000000..ca5643f8
--- /dev/null
+++ b/model/task_plugin_catalog_test.go
@@ -0,0 +1,303 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package model
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/MAX-API-Next/MAX-API/common"
+	"github.com/MAX-API-Next/MAX-API/pkg/taskplugin"
+	"github.com/MAX-API-Next/MAX-API/pkg/taskusage"
+	"github.com/glebarez/sqlite"
+	mysqldriver "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/require"
+	gormmysql "gorm.io/driver/mysql"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+	"gorm.io/gorm/schema"
+)
+
+func catalogFixture(t *testing.T, key, version, source string) (*taskplugin.Package, taskplugin.StoredPackage) {
+	t.Helper()
+	manifest := taskplugin.Manifest{APIVersion: 1, Key: key, Name: key, Version: version, Models: []string{"catalog-model"}, Routes: []taskplugin.Route{{Method: "GET", Path: "/" + key + "/:id", Type: "query", Render: "render"}}}
+	document, err := common.Marshal(taskplugin.PackageDocument{FormatVersion: 1, Manifest: manifest, Source: source})
+	require.NoError(t, err)
+	hash := sha256.Sum256(document)
+	p, err := taskplugin.LoadPackage(bytes.NewReader(document), fmt.Sprintf("%x", hash))
+	require.NoError(t, err)
+	return p, taskplugin.StoredPackage{Document: document, Digest: p.Digest(), Active: true, Enabled: true}
+}
+
+func runTaskPluginCatalogStoreContract(t *testing.T, db *gorm.DB) {
+	t.Helper()
+	ctx := context.Background()
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	store := NewTaskPluginCatalogStore(db)
+	empty, err := store.Load(ctx)
+	require.NoError(t, err)
+	require.Zero(t, empty.Revision)
+	require.Empty(t, empty.Packages)
+	first, entry := catalogFixture(t, "stored", "1.0.0", strings.Repeat("opaque", 15000))
+	require.Greater(t, len(entry.Document), 65535, "exercise MySQL TEXT limit")
+	revision, err := store.CompareAndSwap(ctx, 0, []taskplugin.StoredPackage{entry})
+	require.NoError(t, err)
+	require.EqualValues(t, 1, revision)
+	restarted, err := taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	actor := taskplugin.Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "stored-request"}
+	binding, err := restarted.Host().Bind(actor, "GET", "/stored/one", "catalog-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	loaded, err := store.Load(ctx)
+	require.NoError(t, err)
+	require.Equal(t, entry.Document, loaded.Packages[0].Document)
+	returned := first.Document()
+	returned[0] ^= 1
+	require.Equal(t, entry.Document, first.Document(), "caller must not mutate pinned bytes")
+	_, err = store.CompareAndSwap(ctx, 0, []taskplugin.StoredPackage{entry})
+	require.ErrorIs(t, err, taskplugin.ErrCatalogConflict)
+	second, _ := catalogFixture(t, "stored", "2.0.0", "second")
+	require.NoError(t, restarted.Install(ctx, second))
+	require.NoError(t, restarted.SetEnabled(ctx, "stored", false))
+	fresh, err := taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	_, err = fresh.Host().Bind(actor, "GET", "/stored/one", "catalog-model", taskusage.MiniMaxH3Contract())
+	require.ErrorIs(t, err, taskplugin.ErrBindingRejected)
+	_, err = fresh.Host().Resolve(binding, actor)
+	require.NoError(t, err, "disabled admission must preserve historical tasks")
+	require.NoError(t, fresh.Revoke(ctx, "stored", "1.0.0"))
+	loaded, err = store.Load(ctx)
+	require.NoError(t, err)
+	_, err = store.CompareAndSwap(ctx, loaded.Revision, nil)
+	require.ErrorIs(t, err, taskplugin.ErrCatalogInvalid)
+	for i := range loaded.Packages {
+		loaded.Packages[i].Revoked = false
+	}
+	_, err = store.CompareAndSwap(ctx, loaded.Revision, loaded.Packages)
+	require.ErrorIs(t, err, taskplugin.ErrCatalogInvalid)
+
+	// Fail an INSERT after the head CAS and earlier metadata writes. Both
+	// the revision and all metadata must roll back with the package write.
+	before, err := store.Load(ctx)
+	require.NoError(t, err)
+	_, extra := catalogFixture(t, "extra", "1.0.0", "insert fails")
+	candidate := append(append([]taskplugin.StoredPackage(nil), before.Packages...), extra)
+	for i := range before.Packages {
+		if !candidate[i].Revoked {
+			candidate[i].Enabled = true
+		}
+	}
+	callback := "test_task_plugin_catalog_insert_failure"
+	packageTable := db.NamingStrategy.TableName("TaskPluginVersion")
+	require.NoError(t, db.Callback().Create().Before("gorm:create").Register(callback, func(tx *gorm.DB) {
+		if tx.Statement.Table == packageTable {
+			tx.AddError(errors.New("synthetic package write failure"))
+		}
+	}))
+	_, writeErr := store.CompareAndSwap(ctx, before.Revision, candidate)
+	require.NoError(t, db.Callback().Create().Remove(callback))
+	require.Error(t, writeErr)
+	after, err := store.Load(ctx)
+	require.NoError(t, err)
+	require.Equal(t, before, after)
+
+	final, err := taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	_, err = final.Host().Resolve(binding, actor)
+	require.ErrorIs(t, err, taskplugin.ErrBindingRejected, "revocation survives restart")
+	_, err = final.Host().Bind(actor, "GET", "/stored/one", "catalog-model", taskusage.MiniMaxH3Contract())
+	require.ErrorIs(t, err, taskplugin.ErrBindingRejected)
+	require.NoError(t, final.Activate(ctx, "stored", "2.0.0"))
+	info, _, err := final.List(ctx)
+	require.NoError(t, err)
+	require.Len(t, info, 2)
+}
+
+func TestTaskPluginCatalogStoreSQLite(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "catalog.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	runTaskPluginCatalogStoreContract(t, db)
+}
+
+func runTaskPluginCatalogConcurrentBootstrap(t *testing.T, db *gorm.DB) {
+	t.Helper()
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	_, alpha := catalogFixture(t, "alpha", "1.0.0", "alpha")
+	_, beta := catalogFixture(t, "beta", "1.0.0", "beta")
+	start := make(chan struct{})
+	results := make(chan error, 2)
+	var group sync.WaitGroup
+	for _, entry := range []taskplugin.StoredPackage{alpha, beta} {
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			<-start
+			_, err := NewTaskPluginCatalogStore(db).CompareAndSwap(context.Background(), 0, []taskplugin.StoredPackage{entry})
+			results <- err
+		}()
+	}
+	close(start)
+	group.Wait()
+	close(results)
+	winners := 0
+	for err := range results {
+		if err == nil {
+			winners++
+		}
+	}
+	require.Equal(t, 1, winners)
+	snapshot, err := NewTaskPluginCatalogStore(db).Load(context.Background())
+	require.NoError(t, err)
+	require.EqualValues(t, 1, snapshot.Revision)
+	require.Len(t, snapshot.Packages, 1)
+	// Existing-row CAS must also have one winner; row locking alone must
+	// never allow a stale writer to replace a newer metadata revision.
+	start = make(chan struct{})
+	results = make(chan error, 2)
+	for i := range 2 {
+		entry := snapshot.Packages[0]
+		if i == 0 {
+			entry.Enabled = false
+		} else {
+			entry.Active = false
+		}
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			<-start
+			_, err := NewTaskPluginCatalogStore(db).CompareAndSwap(context.Background(), snapshot.Revision, []taskplugin.StoredPackage{entry})
+			results <- err
+		}()
+	}
+	close(start)
+	group.Wait()
+	close(results)
+	winners = 0
+	for err := range results {
+		if err == nil {
+			winners++
+		}
+	}
+	require.Equal(t, 1, winners)
+	updated, err := NewTaskPluginCatalogStore(db).Load(context.Background())
+	require.NoError(t, err)
+	require.EqualValues(t, 2, updated.Revision)
+}
+
+func TestTaskPluginCatalogSQLiteReopensFile(t *testing.T) {
+	ctx := context.Background()
+	path := filepath.Join(t.TempDir(), "restart.db")
+	open := func() *gorm.DB {
+		db, err := gorm.Open(sqlite.Open(path), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+		require.NoError(t, err)
+		return db
+	}
+	db := open()
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	catalog, err := taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	p, _ := catalogFixture(t, "reopened", "1.0.0", "original")
+	require.NoError(t, catalog.Install(ctx, p))
+	actor := taskplugin.Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "before-close"}
+	binding, err := catalog.Host().Bind(actor, "GET", "/reopened/one", "catalog-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	require.NoError(t, catalog.Revoke(ctx, "reopened", "1.0.0"))
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	require.NoError(t, sqlDB.Close())
+	db = open()
+	sqlDB, err = db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	restarted, err := taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	_, err = restarted.Host().Resolve(binding, actor)
+	require.ErrorIs(t, err, taskplugin.ErrBindingRejected)
+	// Altered stored bytes must prevent a fresh process from opening the host.
+	require.NoError(t, db.Model(&TaskPluginVersion{}).Where("digest = ?", p.Digest()).Update("document", "tampered").Error)
+	_, err = taskplugin.OpenCatalog(ctx, NewTaskPluginCatalogStore(db))
+	require.Error(t, err)
+}
+
+type taskPluginTestDialect struct {
+	gorm.Dialector
+	name string
+}
+
+func (d taskPluginTestDialect) Name() string { return d.name }
+
+func TestTaskPluginCatalogPayloadDialectTypes(t *testing.T) {
+	for dialect, expected := range map[string]string{"mysql": "longtext", "sqlite": "text", "postgres": "text"} {
+		db := &gorm.DB{Config: &gorm.Config{Dialector: taskPluginTestDialect{name: dialect}}}
+		require.Equal(t, expected, taskPluginDocument("").GormDBDataType(db, nil))
+	}
+}
+
+func TestTaskPluginCatalogConcurrentBootstrapSQLite(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "concurrent.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	sqlDB.SetMaxOpenConns(2)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	runTaskPluginCatalogConcurrentBootstrap(t, db)
+}
+
+// Opt-in, isolated two-table fixtures. Existing billing/task/user tables are
+// untouched; cleanup drops only this invocation's random-prefixed tables.
+func TestTaskPluginCatalogMySQL(t *testing.T) {
+	dsn := strings.TrimSpace(os.Getenv("TEST_TASK_PLUGIN_MYSQL_DSN"))
+	if dsn == "" {
+		t.Skip("set TEST_TASK_PLUGIN_MYSQL_DSN for isolated catalog fixtures")
+	}
+	config, err := mysqldriver.ParseDSN(dsn)
+	require.NoError(t, err)
+	allowed := strings.TrimSpace(os.Getenv("TEST_TASK_PLUGIN_MYSQL_DATABASE"))
+	if allowed == "" || config.DBName != allowed {
+		t.Fatal("explicit catalog database authorization required")
+	}
+	_, sqlDB := openMySQLAcceptanceDB(t, dsn)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	for _, name := range []string{"contract", "concurrent"} {
+		t.Run(name, func(t *testing.T) {
+			prefix := "tpc_" + common.GetUUID()[:12] + "_"
+			db, err := gorm.Open(gormmysql.New(gormmysql.Config{Conn: sqlDB, SkipInitializeWithVersion: true}), &gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: prefix}, Logger: logger.Default.LogMode(logger.Silent)})
+			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(&TaskPluginVersion{}, &TaskPluginCatalogHead{})) })
+			if name == "contract" {
+				runTaskPluginCatalogStoreContract(t, db)
+			} else {
+				runTaskPluginCatalogConcurrentBootstrap(t, db)
+			}
+		})
+	}
+}
diff --git a/pkg/taskplugin/catalog.go b/pkg/taskplugin/catalog.go
new file mode 100644
index 00000000..a5c4e730
--- /dev/null
+++ b/pkg/taskplugin/catalog.go
@@ -0,0 +1,353 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package taskplugin
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"sort"
+	"strings"
+	"sync"
+)
+
+var (
+	ErrCatalogConflict    = errors.New("task plugin catalog changed; refresh before retrying")
+	ErrCatalogInvalid     = errors.New("task plugin catalog is invalid")
+	ErrCatalogUnavailable = errors.New("task plugin catalog storage is unavailable")
+)
+
+// StoredPackage contains host-private pinned bytes. Stores must preserve every
+// historical identity, reject changed bytes and make revocation irreversible.
+type StoredPackage struct {
+	Document                 []byte
+	Digest                   string
+	Active, Enabled, Revoked bool
+}
+
+type CatalogSnapshot struct {
+	Revision int64
+	Packages []StoredPackage
+}
+
+// CatalogStore commits a complete version set and the revision in one
+// transaction. A failed CAS must not leave any changed rows behind.
+type CatalogStore interface {
+	Load(context.Context) (CatalogSnapshot, error)
+	CompareAndSwap(context.Context, int64, []StoredPackage) (int64, error)
+}
+
+type PackageInfo struct {
+	Key, Version, Digest     string
+	Active, Enabled, Revoked bool
+}
+
+// Catalog coordinates durable metadata with a single published Host. Disable
+// affects admission only; revoke also invalidates already-issued sessions.
+// Refresh must be called explicitly for changes committed by another node.
+type Catalog struct {
+	mu       sync.Mutex
+	store    CatalogStore
+	host     *Host
+	snapshot CatalogSnapshot
+	state    *hostState
+}
+
+func OpenCatalog(ctx context.Context, store CatalogStore) (*Catalog, error) {
+	if store == nil || ctx == nil {
+		return nil, ErrCatalogUnavailable
+	}
+	catalog := &Catalog{store: store, host: &Host{managed: true}}
+	if err := catalog.Refresh(ctx); err != nil {
+		return nil, err
+	}
+	return catalog, nil
+}
+
+// Host supports binding/restoration, but rejects direct Install/Revoke calls.
+func (catalog *Catalog) Host() *Host { return catalog.host }
+
+func (catalog *Catalog) Refresh(ctx context.Context) error {
+	catalog.mu.Lock()
+	defer catalog.mu.Unlock()
+	return catalog.refreshLocked(ctx)
+}
+
+func (catalog *Catalog) refreshLocked(ctx context.Context) error {
+	if ctx == nil {
+		catalog.host.state.Store(nil)
+		return ErrCatalogUnavailable
+	}
+	snapshot, err := catalog.store.Load(ctx)
+	if err != nil {
+		catalog.host.state.Store(nil)
+		return ErrCatalogUnavailable
+	}
+	state, err := buildCatalogState(snapshot, catalog.snapshot, catalog.state != nil)
+	if err != nil {
+		catalog.host.state.Store(nil)
+		return err
+	}
+	catalog.publish(snapshot, state)
+	return nil
+}
+
+func (catalog *Catalog) publish(snapshot CatalogSnapshot, state *hostState) {
+	catalog.snapshot = cloneCatalogSnapshot(snapshot)
+	catalog.state = state
+	catalog.host.state.Store(state)
+}
+
+// ValidateCatalogTransition is shared with durable stores so even a direct
+// store call cannot rewrite history, clear revocations or publish bad routes.
+func ValidateCatalogTransition(previous, next CatalogSnapshot) error {
+	_, err := buildCatalogState(next, previous, true)
+	return err
+}
+
+func buildCatalogState(snapshot, previous CatalogSnapshot, checkPrevious bool) (*hostState, error) {
+	if snapshot.Revision < 0 || snapshot.Revision == 0 && len(snapshot.Packages) != 0 {
+		return nil, ErrCatalogInvalid
+	}
+	state := &hostState{generation: uint64(snapshot.Revision), active: map[string]*Package{}, history: map[string]*Package{}, revoked: map[string]bool{}}
+	activeKeys := map[string]bool{}
+	total := 0
+	for _, entry := range snapshot.Packages {
+		if len(entry.Document) > MaxPackageBytes || total > MaxRetainedPackageBytes-len(entry.Document) {
+			return nil, ErrCatalogInvalid
+		}
+		total += len(entry.Document)
+		p, err := LoadPackage(bytes.NewReader(entry.Document), entry.Digest)
+		if err != nil || entry.Digest != p.digest {
+			return nil, ErrCatalogInvalid
+		}
+		id := packageIdentity(p.manifest.Key, p.manifest.Version)
+		if state.history[id] != nil {
+			return nil, ErrCatalogInvalid
+		}
+		state.history[id] = p
+		state.revoked[id] = entry.Revoked
+		if entry.Active {
+			if activeKeys[p.manifest.Key] {
+				return nil, ErrCatalogInvalid
+			}
+			activeKeys[p.manifest.Key] = true
+			if entry.Enabled && !entry.Revoked {
+				state.active[p.manifest.Key] = p
+			}
+		}
+	}
+	registry := NewRegistry()
+	keys := make([]string, 0, len(state.active))
+	for key := range state.active {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		if _, err := registry.Publish(state.active[key].manifest); err != nil {
+			return nil, ErrCatalogInvalid
+		}
+	}
+	state.routes = registry.Snapshot()
+	if checkPrevious {
+		if snapshot.Revision < previous.Revision || snapshot.Revision == previous.Revision && catalogFingerprint(snapshot) != catalogFingerprint(previous) {
+			return nil, ErrCatalogInvalid
+		}
+		for _, old := range previous.Packages {
+			p, err := LoadPackage(bytes.NewReader(old.Document), old.Digest)
+			if err != nil {
+				return nil, ErrCatalogInvalid
+			}
+			id := packageIdentity(p.manifest.Key, p.manifest.Version)
+			if current := state.history[id]; current == nil || current.digest != old.Digest || old.Revoked && !state.revoked[id] {
+				return nil, ErrCatalogInvalid
+			}
+		}
+	}
+	return state, nil
+}
+
+func catalogFingerprint(snapshot CatalogSnapshot) string {
+	entries := make([]string, 0, len(snapshot.Packages))
+	for _, entry := range snapshot.Packages {
+		entries = append(entries, fmt.Sprintf("%s:%t:%t:%t", entry.Digest, entry.Active, entry.Enabled, entry.Revoked))
+	}
+	sort.Strings(entries)
+	return strings.Join(entries, "\n")
+}
+
+func cloneCatalogSnapshot(snapshot CatalogSnapshot) CatalogSnapshot {
+	snapshot.Packages = append([]StoredPackage(nil), snapshot.Packages...)
+	for i := range snapshot.Packages {
+		snapshot.Packages[i].Document = append([]byte(nil), snapshot.Packages[i].Document...)
+	}
+	return snapshot
+}
+
+func (catalog *Catalog) update(ctx context.Context, change func(*CatalogSnapshot) error) error {
+	catalog.mu.Lock()
+	defer catalog.mu.Unlock()
+	if err := catalog.refreshLocked(ctx); err != nil {
+		return err
+	}
+	if catalog.snapshot.Revision == math.MaxInt64 {
+		return ErrCatalogInvalid
+	}
+	next := cloneCatalogSnapshot(catalog.snapshot)
+	if err := change(&next); err != nil {
+		return err
+	}
+	next.Revision++
+	state, err := buildCatalogState(next, catalog.snapshot, true)
+	if err != nil {
+		return err
+	}
+	revision, err := catalog.store.CompareAndSwap(ctx, catalog.snapshot.Revision, next.Packages)
+	if err != nil || revision != next.Revision {
+		// A DB error may have happened after commit. Do not publish a guessed
+		// outcome; suspend until Refresh reads the authoritative persisted set.
+		catalog.host.state.Store(nil)
+		if errors.Is(err, ErrCatalogConflict) {
+			return ErrCatalogConflict
+		}
+		return ErrCatalogUnavailable
+	}
+	catalog.publish(next, state)
+	return nil
+}
+
+func (catalog *Catalog) Install(ctx context.Context, packages ...*Package) error {
+	if len(packages) == 0 {
+		return ErrCatalogInvalid
+	}
+	return catalog.update(ctx, func(next *CatalogSnapshot) error {
+		seen := map[string]bool{}
+		for _, p := range packages {
+			if p == nil || seen[p.manifest.Key] {
+				return ErrCatalogInvalid
+			}
+			seen[p.manifest.Key] = true
+			index := -1
+			for i := range next.Packages {
+				old, err := LoadPackage(bytes.NewReader(next.Packages[i].Document), next.Packages[i].Digest)
+				if err != nil {
+					return ErrCatalogInvalid
+				}
+				if old.manifest.Key != p.manifest.Key {
+					continue
+				}
+				if old.manifest.Version == p.manifest.Version {
+					if old.digest != p.digest || next.Packages[i].Revoked {
+						return ErrCatalogInvalid
+					}
+					index = i
+				}
+				next.Packages[i].Active = false
+			}
+			if index < 0 {
+				next.Packages = append(next.Packages, StoredPackage{Document: p.Document(), Digest: p.Digest()})
+				index = len(next.Packages) - 1
+			}
+			next.Packages[index].Active, next.Packages[index].Enabled = true, true
+		}
+		return nil
+	})
+}
+
+func catalogVersion(snapshot *CatalogSnapshot, key, version string, activeOnly bool) (int, error) {
+	for i, entry := range snapshot.Packages {
+		p, err := LoadPackage(bytes.NewReader(entry.Document), entry.Digest)
+		if err != nil {
+			return -1, ErrCatalogInvalid
+		}
+		if p.manifest.Key == key && (activeOnly && entry.Active || !activeOnly && p.manifest.Version == version) {
+			return i, nil
+		}
+	}
+	return -1, ErrBindingRejected
+}
+
+func (catalog *Catalog) Activate(ctx context.Context, key, version string) error {
+	return catalog.update(ctx, func(next *CatalogSnapshot) error {
+		index, err := catalogVersion(next, key, version, false)
+		if err != nil {
+			return err
+		}
+		if next.Packages[index].Revoked {
+			return ErrBindingRejected
+		}
+		for i, entry := range next.Packages {
+			p, _ := LoadPackage(bytes.NewReader(entry.Document), entry.Digest)
+			if p.manifest.Key == key {
+				next.Packages[i].Active = i == index
+			}
+		}
+		next.Packages[index].Enabled = true
+		return nil
+	})
+}
+
+func (catalog *Catalog) SetEnabled(ctx context.Context, key string, enabled bool) error {
+	return catalog.update(ctx, func(next *CatalogSnapshot) error {
+		index, err := catalogVersion(next, key, "", true)
+		if err != nil {
+			return err
+		}
+		if enabled && next.Packages[index].Revoked {
+			return ErrBindingRejected
+		}
+		next.Packages[index].Enabled = enabled
+		return nil
+	})
+}
+
+func (catalog *Catalog) Revoke(ctx context.Context, key, version string) error {
+	return catalog.update(ctx, func(next *CatalogSnapshot) error {
+		index, err := catalogVersion(next, key, version, false)
+		if err != nil {
+			return err
+		}
+		next.Packages[index].Revoked, next.Packages[index].Enabled = true, false
+		return nil
+	})
+}
+
+// List refreshes durable state and returns metadata without source bytes.
+func (catalog *Catalog) List(ctx context.Context) ([]PackageInfo, int64, error) {
+	catalog.mu.Lock()
+	defer catalog.mu.Unlock()
+	if err := catalog.refreshLocked(ctx); err != nil {
+		return nil, 0, err
+	}
+	result := make([]PackageInfo, 0, len(catalog.snapshot.Packages))
+	for _, entry := range catalog.snapshot.Packages {
+		p, _ := LoadPackage(bytes.NewReader(entry.Document), entry.Digest)
+		result = append(result, PackageInfo{Key: p.manifest.Key, Version: p.manifest.Version, Digest: entry.Digest, Active: entry.Active, Enabled: entry.Enabled, Revoked: entry.Revoked})
+	}
+	sort.Slice(result, func(i, j int) bool {
+		if result[i].Key != result[j].Key {
+			return result[i].Key < result[j].Key
+		}
+		return result[i].Version < result[j].Version
+	})
+	return result, catalog.snapshot.Revision, nil
+}
diff --git a/pkg/taskplugin/catalog_test.go b/pkg/taskplugin/catalog_test.go
new file mode 100644
index 00000000..bf04a514
--- /dev/null
+++ b/pkg/taskplugin/catalog_test.go
@@ -0,0 +1,214 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package taskplugin
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/MAX-API-Next/MAX-API/pkg/taskusage"
+	"github.com/stretchr/testify/require"
+)
+
+type memoryCatalogStore struct {
+	sync.Mutex
+	snapshot             CatalogSnapshot
+	failLoad, failCommit bool
+	commitUnknown        bool
+}
+
+func (s *memoryCatalogStore) Load(context.Context) (CatalogSnapshot, error) {
+	s.Lock()
+	defer s.Unlock()
+	if s.failLoad {
+		return CatalogSnapshot{}, errors.New("synthetic load failure")
+	}
+	return cloneCatalogSnapshot(s.snapshot), nil
+}
+
+func (s *memoryCatalogStore) CompareAndSwap(_ context.Context, revision int64, packages []StoredPackage) (int64, error) {
+	s.Lock()
+	defer s.Unlock()
+	if s.failCommit {
+		return 0, errors.New("synthetic commit failure")
+	}
+	if s.snapshot.Revision != revision {
+		return 0, ErrCatalogConflict
+	}
+	s.snapshot = cloneCatalogSnapshot(CatalogSnapshot{Revision: revision + 1, Packages: packages})
+	if s.commitUnknown {
+		return 0, errors.New("synthetic lost commit acknowledgment")
+	}
+	return s.snapshot.Revision, nil
+}
+
+func TestCatalogRestartRetainsVersionsDisablingAndRevocation(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	first := testPackage(t, validManifest(), "first")
+	require.NoError(t, catalog.Install(ctx, first))
+	actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "catalog-task"}
+	binding, err := catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	manifest := validManifest()
+	manifest.Version = "2.0.0"
+	require.NoError(t, catalog.Install(ctx, testPackage(t, manifest, "second")))
+	require.NoError(t, catalog.SetEnabled(ctx, "vendor", false))
+	restarted, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	_, err = restarted.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.ErrorIs(t, err, ErrBindingRejected)
+	session, err := restarted.Host().Resolve(binding, actor)
+	require.NoError(t, err)
+	require.Equal(t, first.Digest(), session.Binding().PackageDigest)
+	require.NoError(t, restarted.Revoke(ctx, "vendor", "1.0.0"))
+	_, err = session.UsageEnvelope(nil)
+	require.ErrorIs(t, err, ErrBindingRejected)
+	third, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	_, err = third.Host().Resolve(binding, actor)
+	require.ErrorIs(t, err, ErrBindingRejected)
+	require.Error(t, third.Activate(ctx, "vendor", "1.0.0"))
+	require.NoError(t, third.Activate(ctx, "vendor", "2.0.0"))
+	_, err = third.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+}
+
+func TestCatalogValidatesBeforeCommitAndFailsClosedOnStorageFailure(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	first := testPackage(t, validManifest(), "original")
+	require.NoError(t, catalog.Install(ctx, first))
+	before := store.snapshot.Revision
+	require.Error(t, catalog.Install(ctx, testPackage(t, registryManifest("conflict", "/vendor/v1/jobs/:id"), "conflict")))
+	require.Equal(t, before, store.snapshot.Revision)
+	require.Error(t, catalog.Install(ctx, testPackage(t, validManifest(), "changed")))
+	require.Equal(t, before, store.snapshot.Revision)
+	_, err = catalog.Host().Install(first)
+	require.Error(t, err, "managed host must not bypass persistence")
+	require.Error(t, catalog.Host().Revoke("vendor", "1.0.0"))
+	store.failCommit = true
+	require.Error(t, catalog.SetEnabled(ctx, "vendor", false))
+	actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "closed-on-error"}
+	_, err = catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.ErrorIs(t, err, ErrBindingRejected)
+	store.failCommit = false
+	require.NoError(t, catalog.Refresh(ctx))
+	_, err = catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	store.failLoad = true
+	require.Error(t, catalog.Refresh(ctx))
+	store.failLoad = false
+	store.snapshot.Packages[0].Document[0] ^= 1
+	_, err = OpenCatalog(ctx, store)
+	require.Error(t, err, "corrupt package must prevent startup")
+}
+
+func TestCatalogRejectsHistoryDeletionRevocationRollbackAndRevisionRollback(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+	require.NoError(t, catalog.Revoke(ctx, "vendor", "1.0.0"))
+	good := cloneCatalogSnapshot(store.snapshot)
+	store.snapshot.Packages[0].Revoked = false
+	store.snapshot.Revision++
+	require.Error(t, catalog.Refresh(ctx))
+	store.snapshot = cloneCatalogSnapshot(good)
+	store.snapshot.Packages = nil
+	store.snapshot.Revision++
+	require.Error(t, catalog.Refresh(ctx))
+	store.snapshot = cloneCatalogSnapshot(good)
+	store.snapshot.Revision--
+	require.Error(t, catalog.Refresh(ctx))
+	store.snapshot = good
+	require.NoError(t, catalog.Refresh(ctx))
+}
+
+func TestCatalogUnknownCommitRecoversOnlyPersistedOutcome(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+	store.commitUnknown = true
+	require.ErrorIs(t, catalog.Revoke(ctx, "vendor", "1.0.0"), ErrCatalogUnavailable)
+	require.Nil(t, catalog.Host().state.Load(), "unknown outcome must suspend the managed host")
+	store.commitUnknown = false
+	require.NoError(t, catalog.Refresh(ctx))
+	info, revision, err := catalog.List(ctx)
+	require.NoError(t, err)
+	require.EqualValues(t, 2, revision)
+	require.True(t, info[0].Revoked)
+	require.Error(t, catalog.SetEnabled(ctx, "vendor", true))
+}
+
+func TestCatalogConcurrentReadersTolerateSuspensionAndRefresh(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+	actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "reader"}
+	binding, err := catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	var group sync.WaitGroup
+	for range 4 {
+		group.Add(1)
+		go func() {
+			defer group.Done()
+			for range 2000 {
+				_, _ = catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+				_, _ = catalog.Host().Resolve(binding, actor)
+			}
+		}()
+	}
+	for range 20 {
+		catalog.host.state.Store(nil)
+		require.NoError(t, catalog.Refresh(ctx))
+	}
+	group.Wait()
+}
+
+func TestCatalogRefreshAppliesAnotherNodesRevocationToExistingSession(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	first, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, first.Install(ctx, testPackage(t, validManifest(), "original")))
+	actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "two-nodes"}
+	binding, err := first.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	session, err := first.Host().Resolve(binding, actor)
+	require.NoError(t, err)
+	second, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, second.Revoke(ctx, "vendor", "1.0.0"))
+	require.NoError(t, first.Refresh(ctx))
+	_, err = session.UsageEnvelope(nil)
+	require.ErrorIs(t, err, ErrBindingRejected)
+}
diff --git a/pkg/taskplugin/host.go b/pkg/taskplugin/host.go
index 5f53248f..1ee87a9b 100644
--- a/pkg/taskplugin/host.go
+++ b/pkg/taskplugin/host.go
@@ -47,7 +47,10 @@ type hostState struct {
 
 // Host atomically publishes a complete route/package generation. Historical
 // packages are retained so an upgrade cannot reinterpret an in-flight task.
-type Host struct{ state atomic.Pointer[hostState] }
+type Host struct {
+	state   atomic.Pointer[hostState]
+	managed bool // immutable after construction; mutations belong to Catalog
+}
 
 const MaxRetainedPackageBytes = 64 << 20
 
@@ -62,7 +65,7 @@ func packageIdentity(key, version string) string { return key + "@" + version }
 // Install publishes a batch all-or-nothing. Reusing a key/version with different
 // bytes is rejected, including after a later version has become active.
 func (host *Host) Install(packages ...*Package) (uint64, error) {
-	if host == nil || host.state.Load() == nil || len(packages) == 0 {
+	if host == nil || host.managed || host.state.Load() == nil || len(packages) == 0 {
 		return 0, errors.New("task plugin host or package batch is empty")
 	}
 	seen := map[string]bool{}
@@ -113,7 +116,7 @@ func (host *Host) Install(packages ...*Package) (uint64, error) {
 // Revoke disables both new bindings and old sessions for one immutable version.
 // A process restart must reapply host-owned revocations before serving traffic.
 func (host *Host) Revoke(key, version string) error {
-	if host == nil || host.state.Load() == nil {
+	if host == nil || host.managed || host.state.Load() == nil {
 		return ErrBindingRejected
 	}
 	id := packageIdentity(key, version)
@@ -155,10 +158,13 @@ func (host *Host) Bind(actor Actor, method, path, model string, contract types.T
 // BindMapped preserves the client's model alias while resolving the declared
 // upstream model. Neither identity may be replaced during later polling.
 func (host *Host) BindMapped(actor Actor, method, path, model, upstreamModel string, contract types.TaskUsageContract) (types.TaskPluginBinding, error) {
-	if host == nil || host.state.Load() == nil || !validActor(actor) {
+	if host == nil || !validActor(actor) {
 		return types.TaskPluginBinding{}, ErrBindingRejected
 	}
 	state := host.state.Load()
+	if state == nil {
+		return types.TaskPluginBinding{}, ErrBindingRejected
+	}
 	match, ok := state.routes.MatchRoute(method, path, upstreamModel)
 	if !ok {
 		return types.TaskPluginBinding{}, ErrBindingRejected
@@ -180,10 +186,13 @@ func (host *Host) BindMapped(actor Actor, method, path, model, upstreamModel str
 // Resolve uses persisted identity, never the currently active version. Missing
 // history or revoked code fails closed; it never retries a provider submission.
 func (host *Host) Resolve(binding types.TaskPluginBinding, actor Actor) (*Session, error) {
-	if host == nil || host.state.Load() == nil || !validActor(actor) || binding.UserID != actor.UserID || binding.TokenID != actor.TokenID || binding.ChannelID != actor.ChannelID || binding.RequestID != actor.RequestID {
+	if host == nil || !validActor(actor) || binding.UserID != actor.UserID || binding.TokenID != actor.TokenID || binding.ChannelID != actor.ChannelID || binding.RequestID != actor.RequestID {
 		return nil, ErrBindingRejected
 	}
 	state := host.state.Load()
+	if state == nil {
+		return nil, ErrBindingRejected
+	}
 	id := packageIdentity(binding.PluginKey, binding.Version)
 	p := state.history[id]
 	if p == nil || state.revoked[id] || binding.PackageDigest != p.digest || binding.Model == "" || binding.Model != strings.TrimSpace(binding.Model) || !declaresModel(p.manifest, binding.UpstreamModel) {
diff --git a/pkg/taskplugin/package.go b/pkg/taskplugin/package.go
index 3c9c8e2e..c3a85d48 100644
--- a/pkg/taskplugin/package.go
+++ b/pkg/taskplugin/package.go
@@ -46,6 +46,7 @@ type Package struct {
 	source   string
 	digest   string
 	size     int
+	document []byte
 }
 
 // LoadPackage accepts exactly one bounded document and requires a trusted hash
@@ -85,8 +86,11 @@ func LoadPackage(reader io.Reader, expectedSHA256 string) (*Package, error) {
 		// and credentials out of any caller's logs and API error response.
 		return nil, errors.New("invalid task plugin package manifest")
 	}
-	return &Package{manifest: cloneManifest(document.Manifest), source: document.Source, digest: hex.EncodeToString(actual[:]), size: len(data)}, nil
+	return &Package{manifest: cloneManifest(document.Manifest), source: document.Source, digest: hex.EncodeToString(actual[:]), size: len(data), document: data}, nil
 }
 
 func (p *Package) Manifest() Manifest { return cloneManifest(p.manifest) }
 func (p *Package) Digest() string     { return p.digest }
+
+// Document returns the exact pinned bytes, not a re-encoded manifest/source.
+func (p *Package) Document() []byte { return append([]byte(nil), p.document...) }
diff --git a/service/task_plugin_catalog.go b/service/task_plugin_catalog.go
new file mode 100644
index 00000000..3576e5b6
--- /dev/null
+++ b/service/task_plugin_catalog.go
@@ -0,0 +1,52 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package service
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/MAX-API-Next/MAX-API/model"
+	"github.com/MAX-API-Next/MAX-API/pkg/taskplugin"
+	"gorm.io/gorm"
+)
+
+var taskPluginCatalog atomic.Pointer[taskplugin.Catalog]
+
+// InitializeTaskPluginCatalog restores all pinned versions and revocations
+// before startup opens traffic. Corrupt/missing storage must stop startup.
+func InitializeTaskPluginCatalog(ctx context.Context, db *gorm.DB) error {
+	catalog, err := taskplugin.OpenCatalog(ctx, model.NewTaskPluginCatalogStore(db))
+	if err != nil {
+		return err
+	}
+	taskPluginCatalog.Store(catalog)
+	return nil
+}
+
+// CurrentTaskPluginCatalog is internal; no public management or relay routes
+// are registered here. Call Refresh to observe other nodes' committed changes.
+func CurrentTaskPluginCatalog() (*taskplugin.Catalog, error) {
+	catalog := taskPluginCatalog.Load()
+	if catalog == nil {
+		return nil, taskplugin.ErrCatalogUnavailable
+	}
+	return catalog, nil
+}
diff --git a/service/task_plugin_catalog_test.go b/service/task_plugin_catalog_test.go
new file mode 100644
index 00000000..c907d546
--- /dev/null
+++ b/service/task_plugin_catalog_test.go
@@ -0,0 +1,71 @@
+/*
+Copyright (C) 2023-2026 MAX-API-Next
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues
+*/
+
+package service
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"path/filepath"
+	"testing"
+
+	"github.com/MAX-API-Next/MAX-API/common"
+	"github.com/MAX-API-Next/MAX-API/model"
+	"github.com/MAX-API-Next/MAX-API/pkg/taskplugin"
+	"github.com/glebarez/sqlite"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+)
+
+func TestTaskPluginCatalogStartupRestoresBeforeExposure(t *testing.T) {
+	previous := taskPluginCatalog.Load()
+	taskPluginCatalog.Store(nil)
+	t.Cleanup(func() { taskPluginCatalog.Store(previous) })
+	_, err := CurrentTaskPluginCatalog()
+	require.ErrorIs(t, err, taskplugin.ErrCatalogUnavailable)
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "startup.db")), &gorm.Config{})
+	require.NoError(t, err)
+	sqlDB, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
+	require.Error(t, InitializeTaskPluginCatalog(context.Background(), db), "missing tables must stop startup")
+	_, err = CurrentTaskPluginCatalog()
+	require.Error(t, err)
+	require.NoError(t, db.AutoMigrate(&model.TaskPluginCatalogHead{}, &model.TaskPluginVersion{}))
+	manifest := taskplugin.Manifest{APIVersion: 1, Key: "startup", Name: "Startup", Version: "1.0.0", Models: []string{"model"}}
+	document, err := common.Marshal(taskplugin.PackageDocument{FormatVersion: 1, Manifest: manifest, Source: "never executed"})
+	require.NoError(t, err)
+	hash := sha256.Sum256(document)
+	p, err := taskplugin.LoadPackage(bytes.NewReader(document), fmt.Sprintf("%x", hash))
+	require.NoError(t, err)
+	seed, err := taskplugin.OpenCatalog(context.Background(), model.NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	require.NoError(t, seed.Install(context.Background(), p))
+	require.NoError(t, seed.Revoke(context.Background(), "startup", "1.0.0"))
+	require.NoError(t, InitializeTaskPluginCatalog(context.Background(), db))
+	catalog, err := CurrentTaskPluginCatalog()
+	require.NoError(t, err)
+	info, revision, err := catalog.List(context.Background())
+	require.NoError(t, err)
+	require.EqualValues(t, 2, revision)
+	require.Len(t, info, 1)
+	require.True(t, info[0].Revoked)
+}

From 6522a16b4b43f6407e9f6d4f4ba13fadae2c0aac Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Mon, 19 Oct 2026 00:20:24 +0800
Subject: [PATCH 2/7] fix(taskplugin): wait for catalog migration and retry
 read conflicts

---
 docs/task-plugin-host.md            |   7 +-
 main.go                             |   5 ++
 model/task_plugin_catalog.go        |  10 +++
 model/task_plugin_catalog_test.go   |   7 +-
 pkg/taskplugin/catalog.go           |  19 ++++-
 pkg/taskplugin/catalog_test.go      | 104 ++++++++++++++++++++++++++++
 service/task_plugin_catalog.go      |  32 +++++++++
 service/task_plugin_catalog_test.go |  88 +++++++++++++++++++++++
 8 files changed, 266 insertions(+), 6 deletions(-)

diff --git a/docs/task-plugin-host.md b/docs/task-plugin-host.md
index 83229136..5f55c269 100644
--- a/docs/task-plugin-host.md
+++ b/docs/task-plugin-host.md
@@ -19,7 +19,7 @@ The service validates task ownership and the frozen usage contract before attach
 
 ## 持久目录与恢复 / Durable catalog and recovery
 
-`taskplugin.OpenCatalog(ctx, model.NewTaskPluginCatalogStore(db))` 加载内部目录，`catalog.Host()` 提供既有绑定与恢复能力。应用启动由 `InitializeTaskPluginCatalog` 完成恢复；缺表、损坏包或无效状态会阻止启动。目录仍没有公开管理路由或自动加载 JS 的行为。
+`taskplugin.OpenCatalog(ctx, model.NewTaskPluginCatalogStore(db))` 加载内部目录，`catalog.Host()` 提供既有绑定与恢复能力。应用启动由 `InitializeTaskPluginCatalog` 完成恢复；损坏包或无效状态会阻止启动。迁移仍由主节点执行，非主节点在恢复前只读等待两张目录表，最长 10 分钟且可取消；等待超时或恢复失败均阻止开放流量。目录仍没有公开管理路由或自动加载 JS 的行为。
 
 - `Install` 批量校验并保存完整包，原始文档字节和摘要保持不变；同 key/version 不允许替换内容。安装版本成为该 key 的启用版本，历史版本保留。
 - `SetEnabled(false)` 停止新任务绑定，旧任务仍能按历史版本恢复。`Activate` 显式切换已有版本，不自动选择“最新”版本。
@@ -27,11 +27,14 @@ The service validates task ownership and the frozen usage contract before attach
 - `task_plugin_catalog_heads` 的版本号条件更新与 `task_plugin_versions` 的变更在同一事务内提交。事务成功后一次发布宿主状态；存储失败、结果未知或刷新失败会暂停宿主，成功 `Refresh` 后恢复。
 - 包文档在 MySQL 使用 LONGTEXT，在 SQLite/PostgreSQL 使用 TEXT；版本身份使用固定长度摘要索引。升级会新增这两张表，备份与恢复应包含两表的一致状态。
 - 其他节点的变更通过显式 `Refresh` 读取。本阶段没有后台同步任务，不保证跨节点实时撤销；后续执行器接入时必须定义刷新时机。目录容量沿用 64 MiB 历史包原始字节上限。
+- 读取快照遇到并发版本冲突时最多尝试 3 次，首次冲突先暂停旧宿主，只有成功读取并验证完整快照才恢复；持续冲突、取消或其他存储错误仍暂停。不自动重试写入、提交结果未知或任务操作。
 
-`OpenCatalog` restores the internal catalog through a GORM store. Startup restores it before serving traffic and stops on missing tables, corrupt bytes or invalid state. Batch installation preserves exact package bytes and immutable key/version identities. Disabling blocks new admission but preserves historical task recovery; activation explicitly selects a retained version. Revocation is permanent and invalidates existing sessions on their next use. Historical versions are never automatically deleted or replaced.
+`OpenCatalog` restores the internal catalog through a GORM store. Startup restores it before serving traffic and stops on corrupt bytes or invalid state. Only the master runs migrations; non-master nodes wait read-only for both catalog tables for at most ten minutes, respecting cancellation. A timeout or failed restoration prevents serving traffic. Batch installation preserves exact package bytes and immutable key/version identities. Disabling blocks new admission but preserves historical task recovery; activation explicitly selects a retained version. Revocation is permanent and invalidates existing sessions on their next use. Historical versions are never automatically deleted or replaced.
 
 A singleton revision CAS serializes metadata transactions across nodes, including concurrent bootstrap. Publication follows commit; uncertain writes or failed refreshes suspend the managed host until a successful refresh. MySQL stores documents as LONGTEXT; SQLite and PostgreSQL use TEXT. Migrations add `task_plugin_catalog_heads` and `task_plugin_versions`; back up and restore both consistently. Other nodes' changes require explicit `Refresh`; background synchronization and immediate multi-node revocation are outside this stage. Retained document bytes remain capped at 64 MiB. The catalog registers no public routes and never evaluates JavaScript.
 
+Snapshot reads retry concurrent revision conflicts for at most three attempts. The first conflict suspends the old host; only a fully read and validated snapshot restores it. Persistent conflicts, cancellation or other storage errors leave it suspended. Writes, ambiguous commits and task operations are never automatically retried.
+
 ## HTTP 能力 / HTTP capability
 
 `Session.NewFetchScope` 只提供 GET：HTTPS、主渠道 origin 与 manifest 额外 host 白名单、禁止环境代理与重定向、每次解析全部 IP 后直连已检查的 IP。回环、私有、链路本地及保留网段均拒绝。渠道 Authorization 只发往精确的渠道 origin，额外 host 不携带该凭据。插件不能提供任意请求头、请求体或 Cookie。
diff --git a/main.go b/main.go
index 7ad55e52..f35a52ba 100644
--- a/main.go
+++ b/main.go
@@ -447,6 +447,11 @@ func InitResources() error {
 	}
 
 	model.CheckSetup()
+	if !common.IsMasterNode {
+		if err := service.WaitForTaskPluginCatalogSchema(context.Background(), model.DB); err != nil {
+			return fmt.Errorf("wait for task plugin catalog schema: %w", err)
+		}
+	}
 	if err := service.InitializeTaskPluginCatalog(context.Background(), model.DB); err != nil {
 		return fmt.Errorf("initialize task plugin catalog: %w", err)
 	}
diff --git a/model/task_plugin_catalog.go b/model/task_plugin_catalog.go
index 7a029b7a..50767f75 100644
--- a/model/task_plugin_catalog.go
+++ b/model/task_plugin_catalog.go
@@ -77,6 +77,16 @@ func NewTaskPluginCatalogStore(db *gorm.DB) *TaskPluginCatalogStore {
 
 var _ taskplugin.CatalogStore = (*TaskPluginCatalogStore)(nil)
 
+// SchemaReady only observes the two catalog tables. Replica startup must not
+// perform DDL; the master remains responsible for migrations.
+func (store *TaskPluginCatalogStore) SchemaReady(ctx context.Context) bool {
+	if store == nil || store.db == nil || ctx == nil || ctx.Err() != nil {
+		return false
+	}
+	db := store.db.WithContext(ctx)
+	return db.Migrator().HasTable(&TaskPluginCatalogHead{}) && db.Migrator().HasTable(&TaskPluginVersion{})
+}
+
 func loadTaskPluginVersions(tx *gorm.DB) ([]taskplugin.StoredPackage, error) {
 	var records []TaskPluginVersion
 	if err := tx.Order("identity").Find(&records).Error; err != nil {
diff --git a/model/task_plugin_catalog_test.go b/model/task_plugin_catalog_test.go
index ca5643f8..df8522e2 100644
--- a/model/task_plugin_catalog_test.go
+++ b/model/task_plugin_catalog_test.go
@@ -57,9 +57,14 @@ func catalogFixture(t *testing.T, key, version, source string) (*taskplugin.Pack
 func runTaskPluginCatalogStoreContract(t *testing.T, db *gorm.DB) {
 	t.Helper()
 	ctx := context.Background()
+	store := NewTaskPluginCatalogStore(db)
+	require.False(t, store.SchemaReady(ctx), "readiness must not migrate missing tables")
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}))
+	require.False(t, store.SchemaReady(ctx), "both catalog tables are required")
 	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
 	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
-	store := NewTaskPluginCatalogStore(db)
+	require.True(t, store.SchemaReady(ctx))
+	require.False(t, store.SchemaReady(nil))
 	empty, err := store.Load(ctx)
 	require.NoError(t, err)
 	require.Zero(t, empty.Revision)
diff --git a/pkg/taskplugin/catalog.go b/pkg/taskplugin/catalog.go
index a5c4e730..a58ffec3 100644
--- a/pkg/taskplugin/catalog.go
+++ b/pkg/taskplugin/catalog.go
@@ -93,12 +93,25 @@ func (catalog *Catalog) Refresh(ctx context.Context) error {
 }
 
 func (catalog *Catalog) refreshLocked(ctx context.Context) error {
-	if ctx == nil {
+	if ctx == nil || ctx.Err() != nil {
 		catalog.host.state.Store(nil)
 		return ErrCatalogUnavailable
 	}
-	snapshot, err := catalog.store.Load(ctx)
-	if err != nil {
+	var snapshot CatalogSnapshot
+	var err error
+	for attempt := 0; attempt < 3; attempt++ {
+		snapshot, err = catalog.store.Load(ctx)
+		if !errors.Is(err, ErrCatalogConflict) {
+			break
+		}
+		// A concurrent commit may revoke a version. Suspend stale sessions
+		// while retrying reads; writes and unknown outcomes are never retried.
+		catalog.host.state.Store(nil)
+		if ctx.Err() != nil {
+			break
+		}
+	}
+	if err != nil || ctx.Err() != nil {
 		catalog.host.state.Store(nil)
 		return ErrCatalogUnavailable
 	}
diff --git a/pkg/taskplugin/catalog_test.go b/pkg/taskplugin/catalog_test.go
index bf04a514..cd101aca 100644
--- a/pkg/taskplugin/catalog_test.go
+++ b/pkg/taskplugin/catalog_test.go
@@ -22,6 +22,7 @@ package taskplugin
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"testing"
 
@@ -34,11 +35,25 @@ type memoryCatalogStore struct {
 	snapshot             CatalogSnapshot
 	failLoad, failCommit bool
 	commitUnknown        bool
+	loadErrors           []error
+	loadCalls            int
+	onLoad               func(int)
 }
 
 func (s *memoryCatalogStore) Load(context.Context) (CatalogSnapshot, error) {
 	s.Lock()
 	defer s.Unlock()
+	s.loadCalls++
+	if s.onLoad != nil {
+		s.onLoad(s.loadCalls)
+	}
+	if len(s.loadErrors) > 0 {
+		err := s.loadErrors[0]
+		s.loadErrors = s.loadErrors[1:]
+		if err != nil {
+			return CatalogSnapshot{}, err
+		}
+	}
 	if s.failLoad {
 		return CatalogSnapshot{}, errors.New("synthetic load failure")
 	}
@@ -212,3 +227,92 @@ func TestCatalogRefreshAppliesAnotherNodesRevocationToExistingSession(t *testing
 	_, err = session.UsageEnvelope(nil)
 	require.ErrorIs(t, err, ErrBindingRejected)
 }
+
+func TestCatalogReadConflictRetriesRestoreAuthoritativeRevocation(t *testing.T) {
+	ctx := context.Background()
+	store := &memoryCatalogStore{}
+	catalog, err := OpenCatalog(ctx, store)
+	require.NoError(t, err)
+	require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+	actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "read-conflict"}
+	binding, err := catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+	require.NoError(t, err)
+	session, err := catalog.Host().Resolve(binding, actor)
+	require.NoError(t, err)
+	store.snapshot.Revision++
+	store.snapshot.Packages[0].Revoked = true
+	store.loadCalls = 0
+	store.loadErrors = []error{ErrCatalogConflict, fmt.Errorf("mixed snapshot: %w", ErrCatalogConflict)}
+	store.onLoad = func(call int) {
+		if call > 1 {
+			require.Nil(t, catalog.Host().state.Load(), "a conflict must suspend stale admission during retries")
+		}
+	}
+	require.NoError(t, catalog.Refresh(ctx))
+	require.Equal(t, 3, store.loadCalls)
+	_, err = session.UsageEnvelope(nil)
+	require.ErrorIs(t, err, ErrBindingRejected, "only the authoritative revoked state may be published")
+}
+
+func TestCatalogReadFailuresAreBoundedAndSuspendExistingSessions(t *testing.T) {
+	for _, test := range []struct {
+		name   string
+		errors []error
+		calls  int
+		cancel bool
+	}{
+		{name: "persistent conflict", errors: []error{ErrCatalogConflict, ErrCatalogConflict, ErrCatalogConflict, nil}, calls: 3},
+		{name: "storage failure", errors: []error{errors.New("storage unavailable"), nil}, calls: 1},
+		{name: "cancelled conflict", errors: []error{ErrCatalogConflict, nil}, calls: 1, cancel: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			store := &memoryCatalogStore{}
+			catalog, err := OpenCatalog(ctx, store)
+			require.NoError(t, err)
+			require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+			actor := Actor{UserID: 1, TokenID: 2, ChannelID: 3, RequestID: "suspended-session"}
+			binding, err := catalog.Host().Bind(actor, "POST", "/vendor/v1/jobs", "vendor-model", taskusage.MiniMaxH3Contract())
+			require.NoError(t, err)
+			session, err := catalog.Host().Resolve(binding, actor)
+			require.NoError(t, err)
+			store.loadCalls = 0
+			store.loadErrors = test.errors
+			if test.cancel {
+				store.onLoad = func(int) { cancel() }
+			}
+			require.ErrorIs(t, catalog.Refresh(ctx), ErrCatalogUnavailable)
+			require.Equal(t, test.calls, store.loadCalls)
+			require.Nil(t, catalog.Host().state.Load())
+			_, err = session.UsageEnvelope(nil)
+			require.ErrorIs(t, err, ErrBindingRejected)
+			_, err = session.NewFetchScope(context.Background(), "https://example.com", "")
+			require.ErrorIs(t, err, ErrBindingRejected)
+		})
+	}
+}
+
+func TestCatalogReadRetriesApplyToListAndMutation(t *testing.T) {
+	for _, operation := range []string{"list", "mutation"} {
+		t.Run(operation, func(t *testing.T) {
+			ctx := context.Background()
+			store := &memoryCatalogStore{}
+			catalog, err := OpenCatalog(ctx, store)
+			require.NoError(t, err)
+			require.NoError(t, catalog.Install(ctx, testPackage(t, validManifest(), "original")))
+			before := store.snapshot.Revision
+			store.loadCalls = 0
+			store.loadErrors = []error{ErrCatalogConflict}
+			if operation == "list" {
+				_, _, err = catalog.List(ctx)
+				require.Equal(t, before, store.snapshot.Revision)
+			} else {
+				err = catalog.SetEnabled(ctx, "vendor", false)
+				require.Equal(t, before+1, store.snapshot.Revision, "retrying a read must not repeat a write")
+			}
+			require.NoError(t, err)
+			require.Equal(t, 2, store.loadCalls)
+		})
+	}
+}
diff --git a/service/task_plugin_catalog.go b/service/task_plugin_catalog.go
index 3576e5b6..f34cb318 100644
--- a/service/task_plugin_catalog.go
+++ b/service/task_plugin_catalog.go
@@ -22,6 +22,7 @@ package service
 import (
 	"context"
 	"sync/atomic"
+	"time"
 
 	"github.com/MAX-API-Next/MAX-API/model"
 	"github.com/MAX-API-Next/MAX-API/pkg/taskplugin"
@@ -30,6 +31,37 @@ import (
 
 var taskPluginCatalog atomic.Pointer[taskplugin.Catalog]
 
+// WaitForTaskPluginCatalogSchema lets non-master nodes wait for the master to
+// create both tables, without DDL or exposing traffic. It waits at most ten
+// minutes and respects an earlier caller deadline or cancellation.
+func WaitForTaskPluginCatalogSchema(ctx context.Context, db *gorm.DB) error {
+	return waitForTaskPluginCatalogSchema(ctx, db, time.Second)
+}
+
+func waitForTaskPluginCatalogSchema(ctx context.Context, db *gorm.DB, interval time.Duration) error {
+	if ctx == nil || db == nil {
+		return taskplugin.ErrCatalogUnavailable
+	}
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+	defer cancel()
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	store := model.NewTaskPluginCatalogStore(db)
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if store.SchemaReady(ctx) {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}
+
 // InitializeTaskPluginCatalog restores all pinned versions and revocations
 // before startup opens traffic. Corrupt/missing storage must stop startup.
 func InitializeTaskPluginCatalog(ctx context.Context, db *gorm.DB) error {
diff --git a/service/task_plugin_catalog_test.go b/service/task_plugin_catalog_test.go
index c907d546..0136e7b1 100644
--- a/service/task_plugin_catalog_test.go
+++ b/service/task_plugin_catalog_test.go
@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/MAX-API-Next/MAX-API/common"
 	"github.com/MAX-API-Next/MAX-API/model"
@@ -33,6 +34,7 @@ import (
 	"github.com/glebarez/sqlite"
 	"github.com/stretchr/testify/require"
 	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
 )
 
 func TestTaskPluginCatalogStartupRestoresBeforeExposure(t *testing.T) {
@@ -69,3 +71,89 @@ func TestTaskPluginCatalogStartupRestoresBeforeExposure(t *testing.T) {
 	require.Len(t, info, 1)
 	require.True(t, info[0].Revoked)
 }
+
+func TestTaskPluginCatalogReplicaWaitsForMasterBeforeExposure(t *testing.T) {
+	previous := taskPluginCatalog.Load()
+	taskPluginCatalog.Store(nil)
+	t.Cleanup(func() { taskPluginCatalog.Store(previous) })
+	path := filepath.Join(t.TempDir(), "replica-startup.db")
+	open := func() *gorm.DB {
+		db, err := gorm.Open(sqlite.Open(path), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+		require.NoError(t, err)
+		pool, err := db.DB()
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, pool.Close()) })
+		return db
+	}
+	master, replica := open(), open()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	finished := make(chan error, 1)
+	go func() {
+		err := waitForTaskPluginCatalogSchema(ctx, replica, time.Millisecond)
+		if err == nil {
+			err = InitializeTaskPluginCatalog(ctx, replica)
+		}
+		finished <- err
+	}()
+	// The replica must observe both tables; it must not create either one.
+	require.False(t, model.NewTaskPluginCatalogStore(replica).SchemaReady(ctx))
+	require.NoError(t, master.AutoMigrate(&model.TaskPluginCatalogHead{}))
+	require.False(t, model.NewTaskPluginCatalogStore(replica).SchemaReady(ctx))
+	_, err := CurrentTaskPluginCatalog()
+	require.ErrorIs(t, err, taskplugin.ErrCatalogUnavailable)
+	select {
+	case err := <-finished:
+		t.Fatalf("replica exposed an incomplete schema: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+	require.NoError(t, master.AutoMigrate(&model.TaskPluginVersion{}))
+	require.NoError(t, <-finished)
+	catalog, err := CurrentTaskPluginCatalog()
+	require.NoError(t, err)
+	info, revision, err := catalog.List(ctx)
+	require.NoError(t, err)
+	require.Empty(t, info)
+	require.Zero(t, revision)
+}
+
+func TestTaskPluginCatalogSchemaWaitRespectsCancellationAndDoesNotMigrate(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "wait.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	pool, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pool.Close()) })
+	for _, mode := range []string{"cancelled", "deadline"} {
+		t.Run(mode, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+			defer cancel()
+			expected := context.DeadlineExceeded
+			if mode == "cancelled" {
+				cancel()
+				expected = context.Canceled
+			}
+			require.ErrorIs(t, waitForTaskPluginCatalogSchema(ctx, db, time.Millisecond), expected)
+			require.False(t, db.Migrator().HasTable(&model.TaskPluginCatalogHead{}))
+			require.False(t, db.Migrator().HasTable(&model.TaskPluginVersion{}))
+		})
+	}
+	require.ErrorIs(t, WaitForTaskPluginCatalogSchema(nil, db), taskplugin.ErrCatalogUnavailable)
+	require.ErrorIs(t, WaitForTaskPluginCatalogSchema(context.Background(), nil), taskplugin.ErrCatalogUnavailable)
+}
+
+func TestTaskPluginCatalogReadySchemaStillRejectsCorruption(t *testing.T) {
+	previous := taskPluginCatalog.Load()
+	taskPluginCatalog.Store(nil)
+	t.Cleanup(func() { taskPluginCatalog.Store(previous) })
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "corrupt.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	pool, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pool.Close()) })
+	require.NoError(t, db.AutoMigrate(&model.TaskPluginCatalogHead{}, &model.TaskPluginVersion{}))
+	require.NoError(t, db.Create(&model.TaskPluginCatalogHead{ID: 1, Revision: 0, MutationID: "corrupt"}).Error)
+	require.NoError(t, WaitForTaskPluginCatalogSchema(context.Background(), db))
+	require.ErrorIs(t, InitializeTaskPluginCatalog(context.Background(), db), taskplugin.ErrCatalogUnavailable)
+	_, err = CurrentTaskPluginCatalog()
+	require.ErrorIs(t, err, taskplugin.ErrCatalogUnavailable)
+}

From 102ec4dd57321900f7b691d3aac0c95a212da227 Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Mon, 19 Oct 2026 01:34:52 +0800
Subject: [PATCH 3/7] fix(taskplugin): classify concurrent catalog bootstrap as
 conflict

---
 model/task_plugin_catalog.go      |  6 +-
 model/task_plugin_catalog_test.go | 92 ++++++++++++++++++++++++++++++-
 2 files changed, 92 insertions(+), 6 deletions(-)

diff --git a/model/task_plugin_catalog.go b/model/task_plugin_catalog.go
index 50767f75..e3147703 100644
--- a/model/task_plugin_catalog.go
+++ b/model/task_plugin_catalog.go
@@ -133,9 +133,6 @@ func (store *TaskPluginCatalogStore) Load(ctx context.Context) (taskplugin.Catal
 		if err != nil {
 			return err
 		}
-		if missing && len(rows) != 0 {
-			return taskplugin.ErrCatalogInvalid
-		}
 		var after TaskPluginCatalogHead
 		err = tx.First(&after, 1).Error
 		// Under READ COMMITTED, reject a mixed snapshot. MySQL's consistent
@@ -144,6 +141,9 @@ func (store *TaskPluginCatalogStore) Load(ctx context.Context) (taskplugin.Catal
 			if !errors.Is(err, gorm.ErrRecordNotFound) {
 				return taskplugin.ErrCatalogConflict
 			}
+			if len(rows) != 0 {
+				return taskplugin.ErrCatalogInvalid
+			}
 		} else if err != nil || head != after {
 			return taskplugin.ErrCatalogConflict
 		}
diff --git a/model/task_plugin_catalog_test.go b/model/task_plugin_catalog_test.go
index df8522e2..a53fcf06 100644
--- a/model/task_plugin_catalog_test.go
+++ b/model/task_plugin_catalog_test.go
@@ -277,6 +277,89 @@ func TestTaskPluginCatalogConcurrentBootstrapSQLite(t *testing.T) {
 	runTaskPluginCatalogConcurrentBootstrap(t, db)
 }
 
+func TestTaskPluginCatalogLoadDistinguishesBootstrapConflictAndOrphans(t *testing.T) {
+	for _, appeared := range []bool{true, false} {
+		t.Run(fmt.Sprintf("head_appeared_%t", appeared), func(t *testing.T) {
+			db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "observations.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+			require.NoError(t, err)
+			pool, err := db.DB()
+			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, pool.Close()) })
+			require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+			_, entry := catalogFixture(t, "observed", "1.0.0", "original")
+			store := NewTaskPluginCatalogStore(db)
+			_, err = store.CompareAndSwap(context.Background(), 0, []taskplugin.StoredPackage{entry})
+			require.NoError(t, err)
+			if !appeared {
+				require.NoError(t, db.Delete(&TaskPluginCatalogHead{}, 1).Error)
+			}
+			headReads := 0
+			callback := "test_catalog_head_observations"
+			headTable := db.NamingStrategy.TableName("TaskPluginCatalogHead")
+			require.NoError(t, db.Callback().Query().After("gorm:query").Register(callback, func(tx *gorm.DB) {
+				if tx.Statement.Table != headTable {
+					return
+				}
+				headReads++
+				if appeared && headReads == 1 {
+					// Supply the first READ COMMITTED observation: the head was
+					// absent, then a complete bootstrap became visible.
+					*tx.Statement.Dest.(*TaskPluginCatalogHead) = TaskPluginCatalogHead{}
+					tx.RowsAffected = 0
+					tx.Error = gorm.ErrRecordNotFound
+				}
+			}))
+			t.Cleanup(func() { require.NoError(t, db.Callback().Query().Remove(callback)) })
+			_, err = store.Load(context.Background())
+			if appeared {
+				require.ErrorIs(t, err, taskplugin.ErrCatalogConflict)
+			} else {
+				require.ErrorIs(t, err, taskplugin.ErrCatalogInvalid, "genuine orphan rows must remain rejected")
+			}
+			require.Equal(t, 2, headReads, "classify the snapshot only after checking whether the head changed")
+		})
+	}
+}
+
+func runTaskPluginCatalogReadCommittedBootstrap(t *testing.T, db *gorm.DB) {
+	t.Helper()
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	_, entry := catalogFixture(t, "read-committed", "1.0.0", "original")
+	installed := false
+	var writerErr error
+	callback := "test_catalog_bootstrap_between_reads"
+	headTable := db.NamingStrategy.TableName("TaskPluginCatalogHead")
+	require.NoError(t, db.Callback().Query().After("gorm:query").Register(callback, func(tx *gorm.DB) {
+		if installed || tx.Statement.Table != headTable || !errors.Is(tx.Error, gorm.ErrRecordNotFound) {
+			return
+		}
+		installed = true
+		// The reader holds a dedicated connection and transaction. Bootstrap
+		// commits on another connection before its version query can start.
+		_, writerErr = NewTaskPluginCatalogStore(db).CompareAndSwap(context.Background(), 0, []taskplugin.StoredPackage{entry})
+	}))
+	t.Cleanup(func() { require.NoError(t, db.Callback().Query().Remove(callback)) })
+	require.NoError(t, db.Connection(func(reader *gorm.DB) error {
+		// Apply only to the next transaction on this connection, leaving the
+		// pool's session isolation unchanged for subsequent tests.
+		if err := reader.Exec("SET TRANSACTION ISOLATION LEVEL READ COMMITTED").Error; err != nil {
+			return err
+		}
+		_, err := NewTaskPluginCatalogStore(reader).Load(context.Background())
+		require.True(t, installed)
+		require.NoError(t, writerErr)
+		require.ErrorIs(t, err, taskplugin.ErrCatalogConflict, "committed bootstrap is a retryable change, not corruption")
+		return nil
+	}))
+	catalog, err := taskplugin.OpenCatalog(context.Background(), NewTaskPluginCatalogStore(db))
+	require.NoError(t, err)
+	info, revision, err := catalog.List(context.Background())
+	require.NoError(t, err)
+	require.EqualValues(t, 1, revision)
+	require.Len(t, info, 1)
+	require.Equal(t, "read-committed", info[0].Key)
+}
+
 // Opt-in, isolated two-table fixtures. Existing billing/task/user tables are
 // untouched; cleanup drops only this invocation's random-prefixed tables.
 func TestTaskPluginCatalogMySQL(t *testing.T) {
@@ -292,16 +375,19 @@ func TestTaskPluginCatalogMySQL(t *testing.T) {
 	}
 	_, sqlDB := openMySQLAcceptanceDB(t, dsn)
 	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
-	for _, name := range []string{"contract", "concurrent"} {
+	for _, name := range []string{"contract", "concurrent", "read_committed_bootstrap"} {
 		t.Run(name, func(t *testing.T) {
 			prefix := "tpc_" + common.GetUUID()[:12] + "_"
 			db, err := gorm.Open(gormmysql.New(gormmysql.Config{Conn: sqlDB, SkipInitializeWithVersion: true}), &gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: prefix}, Logger: logger.Default.LogMode(logger.Silent)})
 			require.NoError(t, err)
 			t.Cleanup(func() { require.NoError(t, db.Migrator().DropTable(&TaskPluginVersion{}, &TaskPluginCatalogHead{})) })
-			if name == "contract" {
+			switch name {
+			case "contract":
 				runTaskPluginCatalogStoreContract(t, db)
-			} else {
+			case "concurrent":
 				runTaskPluginCatalogConcurrentBootstrap(t, db)
+			case "read_committed_bootstrap":
+				runTaskPluginCatalogReadCommittedBootstrap(t, db)
 			}
 		})
 	}

From 2800033879b9188102276d4463e6104a2948785a Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Mon, 19 Oct 2026 02:24:35 +0800
Subject: [PATCH 4/7] fix(taskplugin): preserve pinned bytes across database
 charsets

---
 docs/task-plugin-host.md          |  8 ++-
 model/task_plugin_catalog.go      | 38 +++++++++++++-
 model/task_plugin_catalog_test.go | 82 ++++++++++++++++++++++++++++++-
 3 files changed, 125 insertions(+), 3 deletions(-)

diff --git a/docs/task-plugin-host.md b/docs/task-plugin-host.md
index 5f55c269..4942ccf1 100644
--- a/docs/task-plugin-host.md
+++ b/docs/task-plugin-host.md
@@ -25,14 +25,20 @@ The service validates task ownership and the frozen usage contract before attach
 - `SetEnabled(false)` 停止新任务绑定，旧任务仍能按历史版本恢复。`Activate` 显式切换已有版本，不自动选择“最新”版本。
 - `Revoke` 永久撤销指定版本，新绑定和现有 session 的下一次使用均被拒绝；接口不提供撤销回滚或历史版本删除。
 - `task_plugin_catalog_heads` 的版本号条件更新与 `task_plugin_versions` 的变更在同一事务内提交。事务成功后一次发布宿主状态；存储失败、结果未知或刷新失败会暂停宿主，成功 `Refresh` 后恢复。
-- 包文档在 MySQL 使用 LONGTEXT，在 SQLite/PostgreSQL 使用 TEXT；版本身份使用固定长度摘要索引。升级会新增这两张表，备份与恢复应包含两表的一致状态。
+- 包文档在 MySQL 使用 LONGTEXT，在 SQLite/PostgreSQL 使用 TEXT；新写入采用带版本前缀的 ASCII base64 封装，读取后还原原始字节再校验摘要，避免列字符集改变载荷。旧草稿中的原始 JSON 行仍可读取，不依赖 TEXT 到二进制列的隐式迁移；损坏封装拒绝加载。编码不提供加密或来源认证。版本身份使用固定长度摘要索引，备份与恢复应包含两表的一致状态。
 - 其他节点的变更通过显式 `Refresh` 读取。本阶段没有后台同步任务，不保证跨节点实时撤销；后续执行器接入时必须定义刷新时机。目录容量沿用 64 MiB 历史包原始字节上限。
 - 读取快照遇到并发版本冲突时最多尝试 3 次，首次冲突先暂停旧宿主，只有成功读取并验证完整快照才恢复；持续冲突、取消或其他存储错误仍暂停。不自动重试写入、提交结果未知或任务操作。
 
+64 MiB 容量限制按解码后的原文计算，存储封装有编码体积开销。旧草稿节点不能读取新封装；滚动升级期间应暂停插件安装及新版本写入，所有节点升级完成后再恢复。新编码不能修复已被旧存储破坏的原文，摘要仍会拒绝这些记录；降级应恢复兼容版本与一致备份，不重写摘要来绕过校验。
+
 `OpenCatalog` restores the internal catalog through a GORM store. Startup restores it before serving traffic and stops on corrupt bytes or invalid state. Only the master runs migrations; non-master nodes wait read-only for both catalog tables for at most ten minutes, respecting cancellation. A timeout or failed restoration prevents serving traffic. Batch installation preserves exact package bytes and immutable key/version identities. Disabling blocks new admission but preserves historical task recovery; activation explicitly selects a retained version. Revocation is permanent and invalidates existing sessions on their next use. Historical versions are never automatically deleted or replaced.
 
 A singleton revision CAS serializes metadata transactions across nodes, including concurrent bootstrap. Publication follows commit; uncertain writes or failed refreshes suspend the managed host until a successful refresh. MySQL stores documents as LONGTEXT; SQLite and PostgreSQL use TEXT. Migrations add `task_plugin_catalog_heads` and `task_plugin_versions`; back up and restore both consistently. Other nodes' changes require explicit `Refresh`; background synchronization and immediate multi-node revocation are outside this stage. Retained document bytes remain capped at 64 MiB. The catalog registers no public routes and never evaluates JavaScript.
 
+New documents use a version-prefixed ASCII base64 envelope inside the existing text column. Reads restore the original bytes before digest validation, avoiding character-set-dependent payload changes. Legacy raw JSON rows remain readable without an implicit text-to-binary migration; malformed envelopes fail closed. This encoding provides neither encryption nor source authentication. The 64 MiB retention limit applies to decoded document bytes; stored envelopes add encoding overhead.
+
+Older draft nodes cannot read the new envelope. Pause plugin installation and new-version writes during a rolling upgrade until every node has the new reader. Already damaged legacy documents still fail digest validation; the encoding cannot recover their original bytes. Downgrades require compatible code and a consistent backup, never a rewritten digest that bypasses validation.
+
 Snapshot reads retry concurrent revision conflicts for at most three attempts. The first conflict suspends the old host; only a fully read and validated snapshot restores it. Persistent conflicts, cancellation or other storage errors leave it suspended. Writes, ambiguous commits and task operations are never automatically retried.
 
 ## HTTP 能力 / HTTP capability
diff --git a/model/task_plugin_catalog.go b/model/task_plugin_catalog.go
index e3147703..53863b41 100644
--- a/model/task_plugin_catalog.go
+++ b/model/task_plugin_catalog.go
@@ -23,6 +23,8 @@ import (
 	"bytes"
 	"context"
 	"crypto/sha256"
+	"database/sql/driver"
+	"encoding/base64"
 	"encoding/hex"
 	"errors"
 	"math"
@@ -43,7 +45,41 @@ type TaskPluginCatalogHead struct {
 	MutationID string `gorm:"size:64;not null"`
 }
 
-type taskPluginDocument string
+type taskPluginDocument []byte
+
+const taskPluginDocumentPrefix = "maxapi:taskplugin:base64:v1:"
+
+// Value preserves pinned bytes across database character sets without altering
+// existing text columns. This is an encoding, not encryption or authentication.
+func (document taskPluginDocument) Value() (driver.Value, error) {
+	return taskPluginDocumentPrefix + base64.StdEncoding.EncodeToString(document), nil
+}
+
+// Scan also accepts legacy raw JSON documents. Their digest is still checked
+// by the catalog loader; a malformed versioned envelope must not fall back.
+func (document *taskPluginDocument) Scan(value any) error {
+	var raw []byte
+	switch value := value.(type) {
+	case string:
+		raw = []byte(value)
+	case []byte:
+		raw = bytes.Clone(value)
+	default:
+		return taskplugin.ErrCatalogInvalid
+	}
+	if encoded, ok := bytes.CutPrefix(raw, []byte(taskPluginDocumentPrefix)); ok {
+		if len(encoded) > base64.StdEncoding.EncodedLen(taskplugin.MaxPackageBytes) {
+			return taskplugin.ErrCatalogInvalid
+		}
+		decoded, err := base64.StdEncoding.Strict().DecodeString(string(encoded))
+		if err != nil {
+			return taskplugin.ErrCatalogInvalid
+		}
+		raw = decoded
+	}
+	*document = taskPluginDocument(raw)
+	return nil
+}
 
 func (taskPluginDocument) GormDataType() string { return "string" }
 func (taskPluginDocument) GormDBDataType(db *gorm.DB, _ *schema.Field) string {
diff --git a/model/task_plugin_catalog_test.go b/model/task_plugin_catalog_test.go
index a53fcf06..284404bc 100644
--- a/model/task_plugin_catalog_test.go
+++ b/model/task_plugin_catalog_test.go
@@ -267,6 +267,47 @@ func TestTaskPluginCatalogPayloadDialectTypes(t *testing.T) {
 	}
 }
 
+func TestTaskPluginCatalogDocumentEncodingAndLegacyRoundtrip(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "encoding.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	pool, err := db.DB()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, pool.Close()) })
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	_, entry := catalogFixture(t, "encoding", "1.0.0", "中文🚀😀é\\n\t")
+	store := NewTaskPluginCatalogStore(db)
+	_, err = store.CompareAndSwap(context.Background(), 0, []taskplugin.StoredPackage{entry})
+	require.NoError(t, err)
+	var stored string
+	require.NoError(t, db.Model(&TaskPluginVersion{}).Select("document").Scan(&stored).Error)
+	require.True(t, strings.HasPrefix(stored, taskPluginDocumentPrefix))
+	require.NotRegexp(t, `[^\x00-\x7F]`, stored, "database text contains only ASCII")
+	loaded, err := store.Load(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, entry.Document, loaded.Packages[0].Document)
+	// Preserve old drafts' raw JSON rows and their original digest without
+	// depending on a database-specific implicit text-to-binary conversion.
+	require.NoError(t, db.Model(&TaskPluginVersion{}).Where("digest = ?", entry.Digest).
+		UpdateColumn("document", gorm.Expr("?", string(entry.Document))).Error)
+	require.NoError(t, db.AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	loaded, err = NewTaskPluginCatalogStore(db).Load(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, entry.Document, loaded.Packages[0].Document)
+	require.NoError(t, db.Model(&TaskPluginVersion{}).Where("digest = ?", entry.Digest).
+		UpdateColumn("document", gorm.Expr("?", taskPluginDocumentPrefix+"%%%")).Error)
+	_, err = store.Load(context.Background())
+	require.Error(t, err, "corrupt envelope must never fall back to legacy JSON")
+	for _, value := range []any{nil, 1, taskPluginDocumentPrefix + "Zh==", taskPluginDocumentPrefix + strings.Repeat("A", 2*taskplugin.MaxPackageBytes)} {
+		var document taskPluginDocument
+		require.ErrorIs(t, document.Scan(value), taskplugin.ErrCatalogInvalid)
+	}
+	legacy := []byte(`{"legacy":"bytes"}`)
+	var document taskPluginDocument
+	require.NoError(t, document.Scan(legacy))
+	legacy[0] = '!'
+	require.Equal(t, []byte(`{"legacy":"bytes"}`), []byte(document), "scanner must own the source bytes")
+}
+
 func TestTaskPluginCatalogConcurrentBootstrapSQLite(t *testing.T) {
 	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "concurrent.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
 	require.NoError(t, err)
@@ -360,6 +401,41 @@ func runTaskPluginCatalogReadCommittedBootstrap(t *testing.T, db *gorm.DB) {
 	require.Equal(t, "read-committed", info[0].Key)
 }
 
+func runTaskPluginCatalogCharsetRoundtrip(t *testing.T, db *gorm.DB, charset string) {
+	t.Helper()
+	// MySQL 5.7's utf8 and the startup-supported GBK charset cannot represent
+	// supplementary Unicode. MySQL 8 names three-byte utf8 as utf8mb3.
+	require.Contains(t, []string{"utf8", "gbk"}, charset)
+	require.NoError(t, db.Set("gorm:table_options", "DEFAULT CHARACTER SET "+charset).AutoMigrate(&TaskPluginCatalogHead{}, &TaskPluginVersion{}))
+	var actualCharset string
+	table := db.NamingStrategy.TableName("TaskPluginVersion")
+	require.NoError(t, db.Raw("SELECT CHARACTER_SET_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = 'document'", table).Scan(&actualCharset).Error)
+	if charset == "utf8" {
+		require.Contains(t, []string{"utf8", "utf8mb3"}, actualCharset)
+	} else {
+		require.Equal(t, charset, actualCharset)
+	}
+	_, entry := catalogFixture(t, "unicode", "1.0.0", strings.Repeat("中文🚀😀é\\n", 9000))
+	require.Greater(t, len(entry.Document), 65535)
+	identity, err := taskPluginStorageIdentity(entry)
+	require.NoError(t, err)
+	// Bypass the new Valuer to reproduce the original plain-string write on
+	// this same table. A charset error is distinct from a schema/key failure.
+	err = db.Table(table).Create(map[string]any{"identity": identity, "digest": entry.Digest, "document": string(entry.Document), "active": true, "enabled": true, "revoked": false}).Error
+	var charsetErr *mysqldriver.MySQLError
+	require.ErrorAs(t, err, &charsetErr)
+	// MySQL 8.0.22+ may reject prepared-parameter conversion before INSERT
+	// with ER_IMPOSSIBLE_STRING_CONVERSION instead of incorrect-string1366.
+	require.Contains(t, []uint16{1366, 3988}, charsetErr.Number)
+	store := NewTaskPluginCatalogStore(db)
+	_, err = store.CompareAndSwap(context.Background(), 0, []taskplugin.StoredPackage{entry})
+	require.NoError(t, err, "pinned bytes must roundtrip regardless of the text column's charset")
+	snapshot, err := NewTaskPluginCatalogStore(db).Load(context.Background())
+	require.NoError(t, err)
+	require.Len(t, snapshot.Packages, 1)
+	require.Equal(t, entry.Document, snapshot.Packages[0].Document)
+}
+
 // Opt-in, isolated two-table fixtures. Existing billing/task/user tables are
 // untouched; cleanup drops only this invocation's random-prefixed tables.
 func TestTaskPluginCatalogMySQL(t *testing.T) {
@@ -375,7 +451,7 @@ func TestTaskPluginCatalogMySQL(t *testing.T) {
 	}
 	_, sqlDB := openMySQLAcceptanceDB(t, dsn)
 	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
-	for _, name := range []string{"contract", "concurrent", "read_committed_bootstrap"} {
+	for _, name := range []string{"contract", "concurrent", "read_committed_bootstrap", "charset_roundtrip", "charset_gbk_roundtrip"} {
 		t.Run(name, func(t *testing.T) {
 			prefix := "tpc_" + common.GetUUID()[:12] + "_"
 			db, err := gorm.Open(gormmysql.New(gormmysql.Config{Conn: sqlDB, SkipInitializeWithVersion: true}), &gorm.Config{NamingStrategy: schema.NamingStrategy{TablePrefix: prefix}, Logger: logger.Default.LogMode(logger.Silent)})
@@ -388,6 +464,10 @@ func TestTaskPluginCatalogMySQL(t *testing.T) {
 				runTaskPluginCatalogConcurrentBootstrap(t, db)
 			case "read_committed_bootstrap":
 				runTaskPluginCatalogReadCommittedBootstrap(t, db)
+			case "charset_roundtrip":
+				runTaskPluginCatalogCharsetRoundtrip(t, db, "utf8")
+			case "charset_gbk_roundtrip":
+				runTaskPluginCatalogCharsetRoundtrip(t, db, "gbk")
 			}
 		})
 	}

From 0ec06bcb8c3211454553ded545bf7864fa05a52b Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Mon, 19 Oct 2026 02:52:52 +0800
Subject: [PATCH 5/7] fix(taskplugin): preserve catalog corruption diagnostics

---
 pkg/taskplugin/catalog.go           | 3 +++
 pkg/taskplugin/catalog_test.go      | 9 ++++++++-
 service/task_plugin_catalog_test.go | 2 +-
 3 files changed, 12 insertions(+), 2 deletions(-)

diff --git a/pkg/taskplugin/catalog.go b/pkg/taskplugin/catalog.go
index a58ffec3..bbc6416c 100644
--- a/pkg/taskplugin/catalog.go
+++ b/pkg/taskplugin/catalog.go
@@ -113,6 +113,9 @@ func (catalog *Catalog) refreshLocked(ctx context.Context) error {
 	}
 	if err != nil || ctx.Err() != nil {
 		catalog.host.state.Store(nil)
+		if ctx.Err() == nil && errors.Is(err, ErrCatalogInvalid) {
+			return ErrCatalogInvalid
+		}
 		return ErrCatalogUnavailable
 	}
 	state, err := buildCatalogState(snapshot, catalog.snapshot, catalog.state != nil)
diff --git a/pkg/taskplugin/catalog_test.go b/pkg/taskplugin/catalog_test.go
index cd101aca..ca7a3622 100644
--- a/pkg/taskplugin/catalog_test.go
+++ b/pkg/taskplugin/catalog_test.go
@@ -260,10 +260,13 @@ func TestCatalogReadFailuresAreBoundedAndSuspendExistingSessions(t *testing.T) {
 		errors []error
 		calls  int
 		cancel bool
+		want   error
 	}{
 		{name: "persistent conflict", errors: []error{ErrCatalogConflict, ErrCatalogConflict, ErrCatalogConflict, nil}, calls: 3},
 		{name: "storage failure", errors: []error{errors.New("storage unavailable"), nil}, calls: 1},
 		{name: "cancelled conflict", errors: []error{ErrCatalogConflict, nil}, calls: 1, cancel: true},
+		{name: "corrupt storage", errors: []error{fmt.Errorf("corrupt head: %w", ErrCatalogInvalid)}, calls: 1, want: ErrCatalogInvalid},
+		{name: "cancelled corruption", errors: []error{ErrCatalogInvalid}, calls: 1, cancel: true},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
@@ -282,7 +285,11 @@ func TestCatalogReadFailuresAreBoundedAndSuspendExistingSessions(t *testing.T) {
 			if test.cancel {
 				store.onLoad = func(int) { cancel() }
 			}
-			require.ErrorIs(t, catalog.Refresh(ctx), ErrCatalogUnavailable)
+			want := test.want
+			if want == nil {
+				want = ErrCatalogUnavailable
+			}
+			require.ErrorIs(t, catalog.Refresh(ctx), want)
 			require.Equal(t, test.calls, store.loadCalls)
 			require.Nil(t, catalog.Host().state.Load())
 			_, err = session.UsageEnvelope(nil)
diff --git a/service/task_plugin_catalog_test.go b/service/task_plugin_catalog_test.go
index 0136e7b1..3be5f9c3 100644
--- a/service/task_plugin_catalog_test.go
+++ b/service/task_plugin_catalog_test.go
@@ -153,7 +153,7 @@ func TestTaskPluginCatalogReadySchemaStillRejectsCorruption(t *testing.T) {
 	require.NoError(t, db.AutoMigrate(&model.TaskPluginCatalogHead{}, &model.TaskPluginVersion{}))
 	require.NoError(t, db.Create(&model.TaskPluginCatalogHead{ID: 1, Revision: 0, MutationID: "corrupt"}).Error)
 	require.NoError(t, WaitForTaskPluginCatalogSchema(context.Background(), db))
-	require.ErrorIs(t, InitializeTaskPluginCatalog(context.Background(), db), taskplugin.ErrCatalogUnavailable)
+	require.ErrorIs(t, InitializeTaskPluginCatalog(context.Background(), db), taskplugin.ErrCatalogInvalid)
 	_, err = CurrentTaskPluginCatalog()
 	require.ErrorIs(t, err, taskplugin.ErrCatalogUnavailable)
 }

From 53331fa021ee3f6168624f9293f0de6b358ac1fd Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Tue, 20 Oct 2026 09:20:45 +0800
Subject: [PATCH 6/7] update

---
 .github/workflows/ci.yml | 9 ---------
 1 file changed, 9 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 5e07f454..300585a4 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -78,12 +78,3 @@ jobs:
       - name: Typecheck
         working-directory: web/default
         run: bun run typecheck
-
-      - name: Build
-        working-directory: web/default
-        env:
-          CI: ""
-          DISABLE_ESLINT_PLUGIN: "true"
-          NODE_OPTIONS: "--max-old-space-size=4096"
-          VITE_REACT_APP_VERSION: ${{ github.sha }}
-        run: bun run build

From b2745bbaa9113a11dbe55c03eb4faf0ee213195e Mon Sep 17 00:00:00 2001
From: CSCITech <cscitech@cscitech.top>
Date: Thu, 22 Oct 2026 03:13:32 +0800
Subject: [PATCH 7/7] fix: resolve catalog merge and retain frontend CI build

---
 .github/workflows/ci.yml     |  9 ++++++++
 model/task_plugin_catalog.go | 44 ------------------------------------
 2 files changed, 9 insertions(+), 44 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 521ea66c..4afcfc7e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -80,3 +80,12 @@ jobs:
       - name: Typecheck
         working-directory: web/default
         run: bun run typecheck
+
+      - name: Build
+        working-directory: web/default
+        env:
+          CI: ""
+          DISABLE_ESLINT_PLUGIN: "true"
+          NODE_OPTIONS: "--max-old-space-size=4096"
+          VITE_REACT_APP_VERSION: ${{ github.sha }}
+        run: bun run build
diff --git a/model/task_plugin_catalog.go b/model/task_plugin_catalog.go
index 7442dcf1..463996bd 100644
--- a/model/task_plugin_catalog.go
+++ b/model/task_plugin_catalog.go
@@ -123,14 +123,6 @@ func (store *TaskPluginCatalogStore) SchemaReady(ctx context.Context) bool {
 	return db.Migrator().HasTable(&TaskPluginCatalogHead{}) && db.Migrator().HasTable(&TaskPluginVersion{})
 }
 
-<<<<<<< HEAD
-func loadTaskPluginVersions(tx *gorm.DB) ([]taskplugin.StoredPackage, error) {
-	var records []TaskPluginVersion
-	if err := tx.Order("identity").Find(&records).Error; err != nil {
-		return nil, err
-	}
-	result := make([]taskplugin.StoredPackage, 0, len(records))
-=======
 func loadTaskPluginVersions(tx *gorm.DB) ([]taskplugin.StoredPackage, []string, error) {
 	var records []TaskPluginVersion
 	if err := tx.Order("identity").Find(&records).Error; err != nil {
@@ -138,25 +130,16 @@ func loadTaskPluginVersions(tx *gorm.DB) ([]taskplugin.StoredPackage, []string,
 	}
 	result := make([]taskplugin.StoredPackage, 0, len(records))
 	identities := make([]string, 0, len(records))
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 	for _, record := range records {
 		entry := taskplugin.StoredPackage{Document: []byte(record.Document), Digest: record.Digest, Active: record.Active, Enabled: record.Enabled, Revoked: record.Revoked}
 		identity, err := taskPluginStorageIdentity(entry)
 		if err != nil || identity != record.Identity {
-<<<<<<< HEAD
-			return nil, taskplugin.ErrCatalogInvalid
-		}
-		result = append(result, entry)
-	}
-	return result, nil
-=======
 			return nil, nil, taskplugin.ErrCatalogInvalid
 		}
 		result = append(result, entry)
 		identities = append(identities, identity)
 	}
 	return result, identities, nil
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 }
 
 func taskPluginStorageIdentity(entry taskplugin.StoredPackage) (string, error) {
@@ -184,11 +167,7 @@ func (store *TaskPluginCatalogStore) Load(ctx context.Context) (taskplugin.Catal
 		if !missing && (head.Revision <= 0 || head.MutationID == "") {
 			return taskplugin.ErrCatalogInvalid
 		}
-<<<<<<< HEAD
-		rows, err := loadTaskPluginVersions(tx)
-=======
 		rows, _, err := loadTaskPluginVersions(tx)
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 		if err != nil {
 			return err
 		}
@@ -229,8 +208,6 @@ func (store *TaskPluginCatalogStore) CompareAndSwap(ctx context.Context, revisio
 	if err := taskplugin.ValidateCatalogTransition(taskplugin.CatalogSnapshot{}, next); err != nil {
 		return 0, err
 	}
-<<<<<<< HEAD
-=======
 	// Input identities do not depend on database state. Compute them before
 	// taking the singleton head lock; keep transition validation in the transaction.
 	nextIdentities := make([]string, len(next.Packages))
@@ -241,7 +218,6 @@ func (store *TaskPluginCatalogStore) CompareAndSwap(ctx context.Context, revisio
 		}
 		nextIdentities[i] = identity
 	}
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 	mutationID := common.GetUUID()
 	err := store.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		if revision == 0 {
@@ -266,11 +242,7 @@ func (store *TaskPluginCatalogStore) CompareAndSwap(ctx context.Context, revisio
 				return taskplugin.ErrCatalogConflict
 			}
 		}
-<<<<<<< HEAD
-		previousRows, err := loadTaskPluginVersions(tx)
-=======
 		previousRows, previousIdentities, err := loadTaskPluginVersions(tx)
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 		if err != nil {
 			return err
 		}
@@ -281,28 +253,12 @@ func (store *TaskPluginCatalogStore) CompareAndSwap(ctx context.Context, revisio
 		if err := taskplugin.ValidateCatalogTransition(previous, next); err != nil {
 			return err
 		}
-<<<<<<< HEAD
-		known := make(map[string]bool, len(previousRows))
-		for _, old := range previousRows {
-			identity, err := taskPluginStorageIdentity(old)
-			if err != nil {
-				return err
-			}
-			known[identity] = true
-		}
-		for _, entry := range next.Packages {
-			identity, err := taskPluginStorageIdentity(entry)
-			if err != nil {
-				return err
-			}
-=======
 		known := make(map[string]bool, len(previousIdentities))
 		for _, identity := range previousIdentities {
 			known[identity] = true
 		}
 		for i, entry := range next.Packages {
 			identity := nextIdentities[i]
->>>>>>> b2f25b05258df93a0b2a9935521d6c91cfe6c61f
 			if known[identity] {
 				if err := tx.Model(&TaskPluginVersion{}).Where("identity = ?", identity).
 					Updates(map[string]any{"active": entry.Active, "enabled": entry.Enabled, "revoked": entry.Revoked}).Error; err != nil {