Merge pull request #7351 from Wei-Shaw/feat/plugin-host-services-status-bridge

feat(plugin): generic host services + read-only status bridge channel
2026-10-14 12:02:27 +08:00 · 2026-09-19 11:57:41 +08:00
parent efe9aab1e4 6477143533
commit 4dbcdce43c
22 changed files with 2706 additions and 86 deletions
@@ -204,7 +204,9 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
 	accountUsageService := service.ProvideAccountUsageService(accountRepository, usageLogRepository, claudeUsageFetcher, geminiQuotaService, antigravityQuotaFetcher, grokQuotaFetcher, grokQuotaService, openAIQuotaService, usageCache, identityCache, tlsFingerprintProfileService, openAIGatewayService)
 	pluginRepository := repository.NewPluginRepository(db)
 	pluginHostInfo := providePluginHostInfo(buildInfo)
-	pluginManager := service.NewPluginManager(pluginRepository, secretEncryptor, configConfig, pluginHostInfo)
+	pluginKVStore := repository.NewPluginKVStore(redisClient)
+	pluginManager := service.NewPluginManager(pluginRepository, secretEncryptor, configConfig, pluginHostInfo, pluginKVStore)
+	pluginManager.SetAccountDirectory(openAIGatewayService)
 	accountTestService := service.ProvideAccountTestService(accountRepository, geminiTokenProvider, claudeTokenProvider, grokTokenProvider, antigravityGatewayService, httpUpstream, configConfig, tlsFingerprintProfileService, openAIGatewayService, settingService, pluginManager)
 	crsSyncService := service.NewCRSSyncService(accountRepository, proxyRepository, oAuthService, openAIOAuthService, geminiOAuthService, configConfig)
 	accountHandler := admin.ProvideAccountHandler(configConfig, adminService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, grokOAuthService, rateLimitService, accountUsageService, accountTestService, concurrencyService, crsSyncService, sessionLimitCache, rpmCache, compositeTokenCacheInvalidator, grokQuotaService)
@@ -183,6 +183,22 @@ func (h *PluginHandler) Test(c *gin.Context) {
 	response.Success(c, result)
 }

+// Status returns the plugin's passive runtime status for the config UI. It is
+// read-only (no config apply, no upstream call) and therefore not step-up gated,
+// so a status UI can poll it without a 2FA prompt or a "test" side effect.
+func (h *PluginHandler) Status(c *gin.Context) {
+	id, ok := pluginIDParam(c)
+	if !ok {
+		return
+	}
+	result, err := h.manager.Status(c.Request.Context(), id)
+	if err != nil {
+		response.ErrorFrom(c, err)
+		return
+	}
+	response.Success(c, result)
+}
+
 func (h *PluginHandler) CreateUISession(c *gin.Context) {
 	id, ok := pluginIDParam(c)
 	if !ok {
@@ -0,0 +1,99 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/redis/go-redis/v9"
+)
+
+// pluginKVKeyPrefix 是所有插件通用键值存储在 Redis 中的根前缀。版本段 v1 便于未来
+// 变更内部键结构时平滑迁移。
+const pluginKVKeyPrefix = "plugin:kv:v1:"
+
+// pluginKVScanCount 是 SCAN 每轮的 COUNT 提示，兼顾单轮开销与往返次数。
+const pluginKVScanCount = 200
+
+type pluginKVStore struct {
+	rdb *redis.Client
+}
+
+// NewPluginKVStore 返回 Redis 支撑的通用插件键值存储。使用 Redis 而非进程内存，使得
+// 插件状态可跨副本、跨重启共享，这正是有状态插件所需的通用能力。
+func NewPluginKVStore(rdb *redis.Client) service.PluginKVStore {
+	return &pluginKVStore{rdb: rdb}
+}
+
+// namespacePrefix 拼出某插件某命名空间的内部键前缀。pluginKey / namespace 均已由宿主
+// 服务端按 [A-Za-z0-9._-] 校验，不含 ':' 或 glob 元字符，键结构因此无歧义。
+func (s *pluginKVStore) namespacePrefix(pluginKey, namespace string) string {
+	return pluginKVKeyPrefix + pluginKey + ":" + namespace + ":"
+}
+
+func (s *pluginKVStore) fullKey(pluginKey, namespace, key string) string {
+	return s.namespacePrefix(pluginKey, namespace) + key
+}
+
+func (s *pluginKVStore) Get(ctx context.Context, pluginKey, namespace, key string) ([]byte, bool, error) {
+	value, err := s.rdb.Get(ctx, s.fullKey(pluginKey, namespace, key)).Bytes()
+	if errors.Is(err, redis.Nil) {
+		return nil, false, nil
+	}
+	if err != nil {
+		return nil, false, fmt.Errorf("get plugin kv: %w", err)
+	}
+	return value, true, nil
+}
+
+func (s *pluginKVStore) Set(ctx context.Context, pluginKey, namespace, key string, value []byte, ttl time.Duration) error {
+	// go-redis 将 ttl==0 视为不过期，正好对应“持久保存”语义。
+	if err := s.rdb.Set(ctx, s.fullKey(pluginKey, namespace, key), value, ttl).Err(); err != nil {
+		return fmt.Errorf("set plugin kv: %w", err)
+	}
+	return nil
+}
+
+func (s *pluginKVStore) Delete(ctx context.Context, pluginKey, namespace, key string) error {
+	if err := s.rdb.Del(ctx, s.fullKey(pluginKey, namespace, key)).Err(); err != nil {
+		return fmt.Errorf("delete plugin kv: %w", err)
+	}
+	return nil
+}
+
+func (s *pluginKVStore) List(ctx context.Context, pluginKey, namespace, keyPrefix string, limit int) ([]string, error) {
+	if limit <= 0 {
+		return nil, nil
+	}
+	internalPrefix := s.namespacePrefix(pluginKey, namespace)
+	// keyPrefix 已按 [A-Za-z0-9._-] 校验，不含 glob 元字符，可安全拼接进 MATCH。
+	pattern := internalPrefix + keyPrefix + "*"
+	seen := make(map[string]struct{}, limit)
+	keys := make([]string, 0, limit)
+	var cursor uint64
+	for {
+		batch, next, err := s.rdb.Scan(ctx, cursor, pattern, pluginKVScanCount).Result()
+		if err != nil {
+			return nil, fmt.Errorf("scan plugin kv: %w", err)
+		}
+		for _, full := range batch {
+			// SCAN 可能返回重复键，去重后再裁剪前缀返回面向插件的键名。
+			if _, ok := seen[full]; ok {
+				continue
+			}
+			seen[full] = struct{}{}
+			keys = append(keys, strings.TrimPrefix(full, internalPrefix))
+			if len(keys) >= limit {
+				return keys, nil
+			}
+		}
+		cursor = next
+		if cursor == 0 {
+			break
+		}
+	}
+	return keys, nil
+}
@@ -0,0 +1,148 @@
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestPluginKVStore(t *testing.T) (*pluginKVStore, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	store, ok := NewPluginKVStore(rdb).(*pluginKVStore)
+	require.True(t, ok)
+	return store, mr
+}
+
+func TestPluginKVStore_SetGetDelete(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	ctx := context.Background()
+
+	value, found, err := store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	assert.False(t, found)
+	assert.Nil(t, value)
+
+	require.NoError(t, store.Set(ctx, "p1", "state", "k", []byte("hello"), 0))
+	value, found, err = store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Equal(t, []byte("hello"), value)
+
+	require.NoError(t, store.Delete(ctx, "p1", "state", "k"))
+	_, found, err = store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	assert.False(t, found)
+}
+
+func TestPluginKVStore_TTLExpires(t *testing.T) {
+	store, mr := newTestPluginKVStore(t)
+	ctx := context.Background()
+
+	require.NoError(t, store.Set(ctx, "p1", "state", "k", []byte("v"), 30*time.Second))
+	full := store.fullKey("p1", "state", "k")
+	ttl := mr.TTL(full)
+	assert.Equal(t, 30*time.Second, ttl)
+
+	mr.FastForward(31 * time.Second)
+	_, found, err := store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	assert.False(t, found)
+}
+
+func TestPluginKVStore_PluginIsolation(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	ctx := context.Background()
+
+	require.NoError(t, store.Set(ctx, "p1", "state", "shared", []byte("one"), 0))
+	require.NoError(t, store.Set(ctx, "p2", "state", "shared", []byte("two"), 0))
+
+	v1, found, err := store.Get(ctx, "p1", "state", "shared")
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Equal(t, []byte("one"), v1)
+
+	v2, found, err := store.Get(ctx, "p2", "state", "shared")
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Equal(t, []byte("two"), v2)
+}
+
+func TestPluginKVStore_ListStripsPrefixAndHonorsLimit(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	ctx := context.Background()
+	for _, key := range []string{"account-1", "account-2", "account-3", "other"} {
+		require.NoError(t, store.Set(ctx, "p1", "state", key, []byte("x"), 0))
+	}
+	// 另一命名空间/插件的键不应混入结果。
+	require.NoError(t, store.Set(ctx, "p1", "other-ns", "account-9", []byte("x"), 0))
+	require.NoError(t, store.Set(ctx, "p2", "state", "account-9", []byte("x"), 0))
+
+	keys, err := store.List(ctx, "p1", "state", "account-", 100)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"account-1", "account-2", "account-3"}, keys)
+
+	keys, err = store.List(ctx, "p1", "state", "", 100)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"account-1", "account-2", "account-3", "other"}, keys)
+
+	keys, err = store.List(ctx, "p1", "state", "account-", 2)
+	require.NoError(t, err)
+	assert.Len(t, keys, 2)
+
+	keys, err = store.List(ctx, "p1", "state", "account-", 0)
+	require.NoError(t, err)
+	assert.Empty(t, keys)
+}
+
+// namespace 是另一 namespace 的前缀时（state vs state2），List 不能串味——内部键的 ':'
+// 分隔符 + namespace 字符集排除 ':' 共同保证消歧。
+func TestPluginKVStore_ListNamespacePrefixDisambiguation(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	ctx := context.Background()
+	require.NoError(t, store.Set(ctx, "p1", "state", "k1", []byte("a"), 0))
+	require.NoError(t, store.Set(ctx, "p1", "state2", "k2", []byte("b"), 0))
+
+	keys, err := store.List(ctx, "p1", "state", "", 100)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"k1"}, keys)
+
+	keys, err = store.List(ctx, "p1", "state2", "", 100)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"k2"}, keys)
+}
+
+func TestPluginKVStore_OverwriteResetsValueAndTTL(t *testing.T) {
+	store, mr := newTestPluginKVStore(t)
+	ctx := context.Background()
+	require.NoError(t, store.Set(ctx, "p1", "state", "k", []byte("v1"), 10*time.Second))
+	require.NoError(t, store.Set(ctx, "p1", "state", "k", []byte("v2"), 0))
+
+	value, found, err := store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Equal(t, []byte("v2"), value)
+	// ttl=0 覆盖后应变为不过期。
+	assert.Equal(t, time.Duration(0), mr.TTL(store.fullKey("p1", "state", "k")))
+}
+
+func TestPluginKVStore_EmptyValueIsFound(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	ctx := context.Background()
+	require.NoError(t, store.Set(ctx, "p1", "state", "k", []byte{}, 0))
+	value, found, err := store.Get(ctx, "p1", "state", "k")
+	require.NoError(t, err)
+	assert.True(t, found)
+	assert.Empty(t, value)
+}
+
+func TestPluginKVStore_DeleteMissingIsNoError(t *testing.T) {
+	store, _ := newTestPluginKVStore(t)
+	require.NoError(t, store.Delete(context.Background(), "p1", "state", "missing"))
+}
@@ -97,6 +97,7 @@ var ProviderSet = wire.NewSet(
 	NewErrorPassthroughRepository,
 	NewTLSFingerprintProfileRepository,
 	NewPluginRepository,
+	NewPluginKVStore,
 	NewChannelRepository,
 	NewChannelMonitorRepository,
 	NewChannelMonitorV2Repository,
@@ -759,6 +759,7 @@ func registerPluginRoutes(admin *gin.RouterGroup, h *handler.Handlers, stepUpAut
 		plugins.POST("/:id/disable", gin.HandlerFunc(stepUpAuth), h.Admin.Plugin.Disable)
 		plugins.DELETE("/:id", gin.HandlerFunc(stepUpAuth), h.Admin.Plugin.Delete)
 		plugins.GET("/:id/config", h.Admin.Plugin.GetConfig)
+		plugins.GET("/:id/status", h.Admin.Plugin.Status)
 		plugins.PUT("/:id/config", gin.HandlerFunc(stepUpAuth), h.Admin.Plugin.SaveConfig)
 		plugins.POST("/:id/test", gin.HandlerFunc(stepUpAuth), h.Admin.Plugin.Test)
 		plugins.POST("/:id/ui-session", h.Admin.Plugin.CreateUISession)
@@ -0,0 +1,79 @@
+package service
+
+import (
+	"context"
+	"net/http"
+	"strings"
+)
+
+// OpenAIGatewayService implements service.PluginAccountDirectory for the OpenAI
+// OAuth outbound transport capability. The directory is intentionally scoped to
+// OpenAI OAuth-like, non-shadow accounts regardless of the requested filter, so a
+// plugin can never enumerate or resolve credentials outside that set. The host
+// additionally only wires this directory into plugins whose manifest declares the
+// matching capability (see PluginManager.buildHostServices).
+
+// ListPluginAccounts returns the ids of active OpenAI OAuth-like accounts.
+func (s *OpenAIGatewayService) ListPluginAccounts(ctx context.Context, platform, accountType string) ([]int64, error) {
+	if s == nil || s.accountRepo == nil {
+		return nil, nil
+	}
+	if p := strings.TrimSpace(platform); p != "" && p != PlatformOpenAI {
+		return nil, nil
+	}
+	if at := strings.TrimSpace(accountType); at != "" && at != AccountTypeOAuth {
+		// Setup-token style accounts are OAuth-like but not AccountTypeOAuth; the
+		// current binding only routes AccountTypeOAuth, so honour that filter.
+		return nil, nil
+	}
+	accounts, err := s.accountRepo.ListByPlatform(ctx, PlatformOpenAI)
+	if err != nil {
+		return nil, err
+	}
+	ids := make([]int64, 0, len(accounts))
+	for i := range accounts {
+		account := accounts[i]
+		if account.Status == StatusActive && account.IsOpenAIOAuthLike() && !account.IsShadow() && account.Type == AccountTypeOAuth {
+			ids = append(ids, account.ID)
+		}
+	}
+	return ids, nil
+}
+
+// ResolvePluginOutboundIdentity resolves the access token plus the outbound
+// identity headers and proxy the host would attach to a live request for the
+// account. It returns (nil, nil) for out-of-scope accounts or when no token can
+// be resolved.
+func (s *OpenAIGatewayService) ResolvePluginOutboundIdentity(ctx context.Context, accountID int64) (*PluginOutboundIdentity, error) {
+	if s == nil || s.accountRepo == nil || accountID <= 0 {
+		return nil, nil
+	}
+	account, err := s.accountRepo.GetByID(ctx, accountID)
+	if err != nil {
+		return nil, err
+	}
+	if account == nil || account.Type != AccountTypeOAuth || !account.IsOpenAIOAuthLike() || account.IsShadow() {
+		return nil, nil
+	}
+	token, _, err := s.GetAccessToken(ctx, account)
+	if err != nil {
+		return nil, err
+	}
+	if strings.TrimSpace(token) == "" {
+		return nil, nil
+	}
+	headers := http.Header{}
+	if err := resolveAndSetOpenAIChatGPTAccountHeaders(ctx, s.accountRepo, headers, account); err != nil {
+		return nil, err
+	}
+	ensureCodexIdentityHeaders(headers)
+	enforceCodexIdentityHeaders(headers)
+	return &PluginOutboundIdentity{
+		AccountID:   account.ID,
+		Platform:    account.Platform,
+		AccountType: account.Type,
+		ProxyURL:    resolveAccountProxyURL(account),
+		Token:       token,
+		Headers:     headers,
+	}, nil
+}
@@ -0,0 +1,265 @@
+package service
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	pluginv1 "github.com/Wei-Shaw/sub2api/pkg/pluginapi/v1"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// 通用宿主服务（HostService）的资源上限。这些限制与任何具体插件能力无关，
+// 只用于约束单个插件对共享存储的占用，属于防御性护栏而非业务策略。
+const (
+	// pluginKVMaxPluginKeyLen 与清单 id 的 maxLength 一致（manifest.schema.json），
+	// 确保任何可安装插件的 pluginKey 都能通过校验，不会被误判为不可用。
+	pluginKVMaxPluginKeyLen  = 160
+	pluginKVMaxNamespaceLen  = 128
+	pluginKVMaxKeyLen        = 256
+	pluginKVMaxValueBytes    = 256 * 1024
+	pluginKVMaxListLimit     = 1000
+	pluginKVDefaultListLimit = 100
+	pluginKVMaxTTL           = 90 * 24 * time.Hour
+)
+
+// PluginKVStore 是宿主向插件提供的通用命名空间键值存储端口。它对存储介质保持中立
+// （由 repository 层用 Redis 等实现），供任何插件持久化跨请求 / 跨副本 / 跨重启的
+// 状态。pluginKey 由宿主根据服务该连接的运行时注入，插件无法伪造，从而保证不同
+// 插件之间命名空间严格隔离。
+type PluginKVStore interface {
+	Get(ctx context.Context, pluginKey, namespace, key string) ([]byte, bool, error)
+	Set(ctx context.Context, pluginKey, namespace, key string, value []byte, ttl time.Duration) error
+	Delete(ctx context.Context, pluginKey, namespace, key string) error
+	List(ctx context.Context, pluginKey, namespace, keyPrefix string, limit int) ([]string, error)
+}
+
+// PluginOutboundIdentity 是宿主为某账号解析出的、可直接用于出站请求的身份材料：
+// 访问令牌、宿主会附加的出站请求头，以及账号代理。
+type PluginOutboundIdentity struct {
+	AccountID   int64
+	Platform    string
+	AccountType string
+	ProxyURL    string
+	Token       string
+	Headers     http.Header
+}
+
+// PluginAccountDirectory 让插件枚举其能力所覆盖的账号，并按需解析这些账号的出站身份，
+// 无需等待一条真实请求流经插件。这是一项敏感能力（会把账号凭据交给插件进程），因此
+// 宿主只对「其声明能力确实覆盖这些账号」的插件开放（见 PluginManager.buildHostServices）。
+// 实现方自身也必须把返回范围收敛到该能力对应的账号集合。
+type PluginAccountDirectory interface {
+	ListPluginAccounts(ctx context.Context, platform, accountType string) ([]int64, error)
+	ResolvePluginOutboundIdentity(ctx context.Context, accountID int64) (*PluginOutboundIdentity, error)
+}
+
+// pluginHostServiceServer 实现 pluginv1.HostServiceServer，是宿主经 go-plugin broker
+// 反向暴露给单个插件进程的服务端点。它绑定到具体插件的 pluginKey，因此每个运行时都有
+// 自己的实例；所有键值操作都被强制限定在该插件的命名空间内。
+type pluginHostServiceServer struct {
+	pluginv1.UnimplementedHostServiceServer
+	pluginKey string
+	store     PluginKVStore
+	directory PluginAccountDirectory
+}
+
+func newPluginHostServiceServer(pluginKey string, store PluginKVStore, directory PluginAccountDirectory) *pluginHostServiceServer {
+	return &pluginHostServiceServer{pluginKey: pluginKey, store: store, directory: directory}
+}
+
+func (s *pluginHostServiceServer) ready() bool {
+	return s != nil && s.store != nil && isValidPluginKVSegment(s.pluginKey, pluginKVMaxPluginKeyLen)
+}
+
+func (s *pluginHostServiceServer) KVGet(ctx context.Context, req *pluginv1.KVGetRequest) (*pluginv1.KVGetResponse, error) {
+	if !s.ready() {
+		return nil, status.Error(codes.Unavailable, "宿主键值存储不可用")
+	}
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "请求为空")
+	}
+	if err := validatePluginKVNamespace(req.Namespace); err != nil {
+		return nil, err
+	}
+	if err := validatePluginKVKey(req.Key); err != nil {
+		return nil, err
+	}
+	value, found, err := s.store.Get(ctx, s.pluginKey, req.Namespace, req.Key)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "读取键值失败: %v", err)
+	}
+	if !found {
+		return &pluginv1.KVGetResponse{Found: false}, nil
+	}
+	return &pluginv1.KVGetResponse{Found: true, Value: value}, nil
+}
+
+func (s *pluginHostServiceServer) KVSet(ctx context.Context, req *pluginv1.KVSetRequest) (*pluginv1.KVSetResponse, error) {
+	if !s.ready() {
+		return nil, status.Error(codes.Unavailable, "宿主键值存储不可用")
+	}
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "请求为空")
+	}
+	if err := validatePluginKVNamespace(req.Namespace); err != nil {
+		return nil, err
+	}
+	if err := validatePluginKVKey(req.Key); err != nil {
+		return nil, err
+	}
+	if len(req.Value) > pluginKVMaxValueBytes {
+		return nil, status.Errorf(codes.InvalidArgument, "值超过 %d 字节上限", pluginKVMaxValueBytes)
+	}
+	ttl, err := pluginKVTTL(req.TtlSeconds)
+	if err != nil {
+		return nil, err
+	}
+	if err := s.store.Set(ctx, s.pluginKey, req.Namespace, req.Key, req.Value, ttl); err != nil {
+		return nil, status.Errorf(codes.Internal, "写入键值失败: %v", err)
+	}
+	return &pluginv1.KVSetResponse{}, nil
+}
+
+func (s *pluginHostServiceServer) KVDelete(ctx context.Context, req *pluginv1.KVDeleteRequest) (*pluginv1.KVDeleteResponse, error) {
+	if !s.ready() {
+		return nil, status.Error(codes.Unavailable, "宿主键值存储不可用")
+	}
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "请求为空")
+	}
+	if err := validatePluginKVNamespace(req.Namespace); err != nil {
+		return nil, err
+	}
+	if err := validatePluginKVKey(req.Key); err != nil {
+		return nil, err
+	}
+	if err := s.store.Delete(ctx, s.pluginKey, req.Namespace, req.Key); err != nil {
+		return nil, status.Errorf(codes.Internal, "删除键值失败: %v", err)
+	}
+	return &pluginv1.KVDeleteResponse{}, nil
+}
+
+func (s *pluginHostServiceServer) KVList(ctx context.Context, req *pluginv1.KVListRequest) (*pluginv1.KVListResponse, error) {
+	if !s.ready() {
+		return nil, status.Error(codes.Unavailable, "宿主键值存储不可用")
+	}
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "请求为空")
+	}
+	if err := validatePluginKVNamespace(req.Namespace); err != nil {
+		return nil, err
+	}
+	if req.KeyPrefix != "" {
+		if err := validatePluginKVKey(req.KeyPrefix); err != nil {
+			return nil, err
+		}
+	}
+	limit := int(req.Limit)
+	if limit <= 0 {
+		limit = pluginKVDefaultListLimit
+	}
+	if limit > pluginKVMaxListLimit {
+		limit = pluginKVMaxListLimit
+	}
+	keys, err := s.store.List(ctx, s.pluginKey, req.Namespace, req.KeyPrefix, limit)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "列举键值失败: %v", err)
+	}
+	return &pluginv1.KVListResponse{Keys: keys}, nil
+}
+
+func (s *pluginHostServiceServer) ListAccounts(ctx context.Context, req *pluginv1.ListAccountsRequest) (*pluginv1.ListAccountsResponse, error) {
+	if s == nil || s.directory == nil {
+		return nil, status.Error(codes.Unavailable, "账号目录不可用")
+	}
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "请求为空")
+	}
+	ids, err := s.directory.ListPluginAccounts(ctx, req.Platform, req.AccountType)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "列举账号失败: %v", err)
+	}
+	return &pluginv1.ListAccountsResponse{AccountIds: ids}, nil
+}
+
+func (s *pluginHostServiceServer) ResolveOutboundIdentity(ctx context.Context, req *pluginv1.ResolveOutboundIdentityRequest) (*pluginv1.ResolveOutboundIdentityResponse, error) {
+	if s == nil || s.directory == nil {
+		return nil, status.Error(codes.Unavailable, "账号目录不可用")
+	}
+	if req == nil || req.AccountId <= 0 {
+		return nil, status.Error(codes.InvalidArgument, "account_id 无效")
+	}
+	identity, err := s.directory.ResolvePluginOutboundIdentity(ctx, req.AccountId)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "解析账号出站身份失败: %v", err)
+	}
+	if identity == nil {
+		return &pluginv1.ResolveOutboundIdentityResponse{Found: false}, nil
+	}
+	return &pluginv1.ResolveOutboundIdentityResponse{
+		Found:       true,
+		AccountId:   identity.AccountID,
+		Platform:    identity.Platform,
+		AccountType: identity.AccountType,
+		ProxyUrl:    identity.ProxyURL,
+		Token:       identity.Token,
+		Headers:     headersToPlugin(identity.Headers),
+	}, nil
+}
+
+func pluginKVTTL(seconds int64) (time.Duration, error) {
+	if seconds < 0 {
+		return 0, status.Error(codes.InvalidArgument, "ttl_seconds 不能为负")
+	}
+	if seconds == 0 {
+		return 0, nil
+	}
+	// 先在整数秒上比较上限，避免 seconds*time.Second 溢出 int64 后回绕成负值、
+	// 从而绕过上限检查把负 TTL 传给 Redis。
+	maxSeconds := int64(pluginKVMaxTTL / time.Second)
+	if seconds > maxSeconds {
+		return 0, status.Errorf(codes.InvalidArgument, "ttl_seconds 超过上限 %d", maxSeconds)
+	}
+	return time.Duration(seconds) * time.Second, nil
+}
+
+func validatePluginKVNamespace(namespace string) error {
+	if namespace == "" {
+		return status.Error(codes.InvalidArgument, "namespace 不能为空")
+	}
+	if !isValidPluginKVSegment(namespace, pluginKVMaxNamespaceLen) {
+		return status.Error(codes.InvalidArgument, "namespace 仅允许字母、数字、'.'、'_'、'-' 且长度受限")
+	}
+	return nil
+}
+
+func validatePluginKVKey(key string) error {
+	if key == "" {
+		return status.Error(codes.InvalidArgument, "key 不能为空")
+	}
+	if !isValidPluginKVSegment(key, pluginKVMaxKeyLen) {
+		return status.Error(codes.InvalidArgument, "key 仅允许字母、数字、'.'、'_'、'-' 且长度受限")
+	}
+	return nil
+}
+
+// isValidPluginKVSegment 限定命名段字符集为 [A-Za-z0-9._-]。这既避免 Redis glob
+// 元字符（*?[]）污染 SCAN，也避免 ':' 破坏内部键结构，从而保证命名空间隔离可靠。
+func isValidPluginKVSegment(value string, maxLen int) bool {
+	if value == "" || len(value) > maxLen {
+		return false
+	}
+	for _, r := range value {
+		switch {
+		case r >= 'a' && r <= 'z':
+		case r >= 'A' && r <= 'Z':
+		case r >= '0' && r <= '9':
+		case r == '.' || r == '_' || r == '-':
+		default:
+			return false
+		}
+	}
+	return true
+}
@@ -0,0 +1,130 @@
+package service
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	pluginv1 "github.com/Wei-Shaw/sub2api/pkg/pluginapi/v1"
+	hcplugin "github.com/hashicorp/go-plugin"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// brokerProbePlugin 是仅用于测试的插件传输实现：它实现 HostBrokerReceiver 以拿到
+// broker，并在 InitHostServices 中用宿主下发的 broker id 拨号回宿主的 HostService，
+// 走一次真实 gRPC 的 KVSet/KVGet 往返。
+type brokerProbePlugin struct {
+	pluginv1.UnimplementedTransportPluginServer
+	broker *hcplugin.GRPCBroker
+
+	mu       sync.Mutex
+	ready    bool
+	getFound bool
+	getValue []byte
+	failure  string
+}
+
+func (p *brokerProbePlugin) SetHostBroker(broker *hcplugin.GRPCBroker) { p.broker = broker }
+
+func (p *brokerProbePlugin) InitHostServices(ctx context.Context, req *pluginv1.InitHostServicesRequest) (*pluginv1.InitHostServicesResponse, error) {
+	if p.broker == nil {
+		return &pluginv1.InitHostServicesResponse{Ready: false, Message: "broker 未注入"}, nil
+	}
+	conn, err := p.broker.Dial(req.HostServiceId)
+	if err != nil {
+		p.record("dial: " + err.Error())
+		return &pluginv1.InitHostServicesResponse{Ready: false, Message: err.Error()}, nil
+	}
+	defer func() { _ = conn.Close() }()
+	client := pluginv1.NewHostServiceClient(conn)
+	if _, err := client.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "probe", Value: []byte("hello"), TtlSeconds: 60}); err != nil {
+		p.record("set: " + err.Error())
+		return &pluginv1.InitHostServicesResponse{Ready: false, Message: err.Error()}, nil
+	}
+	got, err := client.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "probe"})
+	if err != nil {
+		p.record("get: " + err.Error())
+		return &pluginv1.InitHostServicesResponse{Ready: false, Message: err.Error()}, nil
+	}
+	p.mu.Lock()
+	p.ready = true
+	p.getFound = got.Found
+	p.getValue = got.Value
+	p.mu.Unlock()
+	return &pluginv1.InitHostServicesResponse{Ready: true}, nil
+}
+
+func (p *brokerProbePlugin) record(failure string) {
+	p.mu.Lock()
+	p.failure = failure
+	p.mu.Unlock()
+}
+
+// noHostServicesPlugin 模拟老插件：不实现 InitHostServices（返回 Unimplemented），
+// 也不接收 broker。
+type noHostServicesPlugin struct {
+	pluginv1.UnimplementedTransportPluginServer
+}
+
+func dispenseTransportClient(t *testing.T, impl pluginv1.TransportPluginServer) *pluginv1.TransportClient {
+	t.Helper()
+	client, _ := hcplugin.TestPluginGRPCConn(t, false, map[string]hcplugin.Plugin{
+		pluginv1.TransportPluginName: &pluginv1.GRPCPlugin{Impl: impl},
+	})
+	t.Cleanup(func() { _ = client.Close() })
+	dispensed, err := client.Dispense(pluginv1.TransportPluginName)
+	require.NoError(t, err)
+	tc, ok := dispensed.(*pluginv1.TransportClient)
+	require.True(t, ok)
+	require.NotNil(t, tc.Broker)
+	return tc
+}
+
+// 端到端验证 broker 反向通道：宿主服务 AcceptAndServe + 插件 Dial 回来，经真实 gRPC
+// 完成 KV 往返，且写入落到宿主注入的 pluginKey 命名空间下。
+func TestOfferPluginHostServices_BrokerRoundtrip(t *testing.T) {
+	probe := &brokerProbePlugin{}
+	tc := dispenseTransportClient(t, probe)
+
+	store := newFakePluginKVStore()
+	hostServer := newPluginHostServiceServer("test.plugin", store, nil)
+	offerPluginHostServices(context.Background(), &PluginInstallation{PluginKey: "test.plugin"}, tc.TransportPluginClient, tc.Broker, hostServer, 5*time.Second)
+
+	probe.mu.Lock()
+	defer probe.mu.Unlock()
+	require.Empty(t, probe.failure)
+	require.True(t, probe.ready)
+	assert.True(t, probe.getFound)
+	assert.Equal(t, []byte("hello"), probe.getValue)
+
+	stored, found, err := store.Get(context.Background(), "test.plugin", "state", "probe")
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Equal(t, []byte("hello"), stored)
+}
+
+// 老插件不实现 InitHostServices 时，offerPluginHostServices 必须优雅降级、不 panic、
+// 不写入任何状态。
+func TestOfferPluginHostServices_UnimplementedIsGraceful(t *testing.T) {
+	tc := dispenseTransportClient(t, &noHostServicesPlugin{})
+
+	store := newFakePluginKVStore()
+	hostServer := newPluginHostServiceServer("test.plugin", store, nil)
+	require.NotPanics(t, func() {
+		offerPluginHostServices(context.Background(), &PluginInstallation{PluginKey: "test.plugin"}, tc.TransportPluginClient, tc.Broker, hostServer, 5*time.Second)
+	})
+
+	_, found, err := store.Get(context.Background(), "test.plugin", "state", "probe")
+	require.NoError(t, err)
+	assert.False(t, found)
+}
+
+// hostServices 为 nil（未配置键值存储）时不得触发任何 broker 交互。
+func TestOfferPluginHostServices_NilHostServicesNoop(t *testing.T) {
+	tc := dispenseTransportClient(t, &noHostServicesPlugin{})
+	require.NotPanics(t, func() {
+		offerPluginHostServices(context.Background(), &PluginInstallation{PluginKey: "test.plugin"}, tc.TransportPluginClient, tc.Broker, nil, 5*time.Second)
+	})
+}
@@ -0,0 +1,331 @@
+package service
+
+import (
+	"context"
+	"math"
+	"net/http"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	pluginv1 "github.com/Wei-Shaw/sub2api/pkg/pluginapi/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// fakePluginKVStore 是内存版 PluginKVStore，按 (pluginKey, namespace, key) 三元组隔离，
+// 用于确定性地验证宿主服务端的校验与命名空间隔离逻辑。
+type fakePluginKVStore struct {
+	mu     sync.Mutex
+	values map[string][]byte
+	ttls   map[string]time.Duration
+}
+
+func newFakePluginKVStore() *fakePluginKVStore {
+	return &fakePluginKVStore{values: map[string][]byte{}, ttls: map[string]time.Duration{}}
+}
+
+func (f *fakePluginKVStore) compose(pluginKey, namespace, key string) string {
+	return pluginKey + "\x00" + namespace + "\x00" + key
+}
+
+func (f *fakePluginKVStore) Get(_ context.Context, pluginKey, namespace, key string) ([]byte, bool, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	value, ok := f.values[f.compose(pluginKey, namespace, key)]
+	if !ok {
+		return nil, false, nil
+	}
+	return append([]byte(nil), value...), true, nil
+}
+
+func (f *fakePluginKVStore) Set(_ context.Context, pluginKey, namespace, key string, value []byte, ttl time.Duration) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	composed := f.compose(pluginKey, namespace, key)
+	f.values[composed] = append([]byte(nil), value...)
+	f.ttls[composed] = ttl
+	return nil
+}
+
+func (f *fakePluginKVStore) Delete(_ context.Context, pluginKey, namespace, key string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	composed := f.compose(pluginKey, namespace, key)
+	delete(f.values, composed)
+	delete(f.ttls, composed)
+	return nil
+}
+
+func (f *fakePluginKVStore) List(_ context.Context, pluginKey, namespace, keyPrefix string, limit int) ([]string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	prefix := f.compose(pluginKey, namespace, keyPrefix)
+	scope := f.compose(pluginKey, namespace, "")
+	var keys []string
+	for composed := range f.values {
+		if strings.HasPrefix(composed, prefix) {
+			keys = append(keys, strings.TrimPrefix(composed, scope))
+		}
+	}
+	sort.Strings(keys)
+	if limit > 0 && len(keys) > limit {
+		keys = keys[:limit]
+	}
+	return keys, nil
+}
+
+func TestPluginHostServiceServer_SetGetDeleteRoundtrip(t *testing.T) {
+	store := newFakePluginKVStore()
+	server := newPluginHostServiceServer("local.example.plugin", store, nil)
+	ctx := context.Background()
+
+	_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "alpha", Value: []byte("v1"), TtlSeconds: 60})
+	require.NoError(t, err)
+
+	got, err := server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "alpha"})
+	require.NoError(t, err)
+	assert.True(t, got.Found)
+	assert.Equal(t, []byte("v1"), got.Value)
+	assert.Equal(t, time.Minute, store.ttls[store.compose("local.example.plugin", "state", "alpha")])
+
+	_, err = server.KVDelete(ctx, &pluginv1.KVDeleteRequest{Namespace: "state", Key: "alpha"})
+	require.NoError(t, err)
+
+	got, err = server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "alpha"})
+	require.NoError(t, err)
+	assert.False(t, got.Found)
+}
+
+func TestPluginHostServiceServer_GetNotFound(t *testing.T) {
+	server := newPluginHostServiceServer("local.example.plugin", newFakePluginKVStore(), nil)
+	got, err := server.KVGet(context.Background(), &pluginv1.KVGetRequest{Namespace: "state", Key: "missing"})
+	require.NoError(t, err)
+	assert.False(t, got.Found)
+	assert.Nil(t, got.Value)
+}
+
+func TestPluginHostServiceServer_NamespaceIsolationAcrossPlugins(t *testing.T) {
+	store := newFakePluginKVStore()
+	first := newPluginHostServiceServer("local.plugin.one", store, nil)
+	second := newPluginHostServiceServer("local.plugin.two", store, nil)
+	ctx := context.Background()
+
+	_, err := first.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "shared", Value: []byte("first")})
+	require.NoError(t, err)
+
+	// 另一个插件即便使用完全相同的 namespace/key 也读不到，因为 pluginKey 由宿主注入。
+	got, err := second.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "shared"})
+	require.NoError(t, err)
+	assert.False(t, got.Found)
+
+	got, err = first.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "shared"})
+	require.NoError(t, err)
+	assert.True(t, got.Found)
+	assert.Equal(t, []byte("first"), got.Value)
+}
+
+func TestPluginHostServiceServer_ListPrefixAndLimit(t *testing.T) {
+	store := newFakePluginKVStore()
+	server := newPluginHostServiceServer("local.example.plugin", store, nil)
+	ctx := context.Background()
+	for _, key := range []string{"account-1", "account-2", "account-3", "other-1"} {
+		_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: key, Value: []byte("x")})
+		require.NoError(t, err)
+	}
+
+	resp, err := server.KVList(ctx, &pluginv1.KVListRequest{Namespace: "state", KeyPrefix: "account-"})
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"account-1", "account-2", "account-3"}, resp.Keys)
+
+	resp, err = server.KVList(ctx, &pluginv1.KVListRequest{Namespace: "state", KeyPrefix: "account-", Limit: 2})
+	require.NoError(t, err)
+	assert.Len(t, resp.Keys, 2)
+}
+
+func TestPluginHostServiceServer_Validation(t *testing.T) {
+	server := newPluginHostServiceServer("local.example.plugin", newFakePluginKVStore(), nil)
+	ctx := context.Background()
+	oversized := make([]byte, pluginKVMaxValueBytes+1)
+
+	cases := []struct {
+		name string
+		call func() error
+	}{
+		{"empty namespace", func() error {
+			_, err := server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "", Key: "k"})
+			return err
+		}},
+		{"namespace with colon", func() error {
+			_, err := server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "a:b", Key: "k"})
+			return err
+		}},
+		{"key with glob", func() error {
+			_, err := server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: "a*"})
+			return err
+		}},
+		{"empty key", func() error {
+			_, err := server.KVGet(ctx, &pluginv1.KVGetRequest{Namespace: "state", Key: ""})
+			return err
+		}},
+		{"oversized value", func() error {
+			_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: oversized})
+			return err
+		}},
+		{"negative ttl", func() error {
+			_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: []byte("v"), TtlSeconds: -1})
+			return err
+		}},
+		{"ttl over max", func() error {
+			_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: []byte("v"), TtlSeconds: int64(pluginKVMaxTTL/time.Second) + 1})
+			return err
+		}},
+		{"ttl overflow", func() error {
+			// 溢出用例：seconds*time.Second 会回绕成负值，必须仍判为超限而非放行。
+			_, err := server.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: []byte("v"), TtlSeconds: math.MaxInt64})
+			return err
+		}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.call()
+			require.Error(t, err)
+			assert.Equal(t, codes.InvalidArgument, status.Code(err))
+		})
+	}
+}
+
+func TestPluginHostServiceServer_UnavailableWhenNoStore(t *testing.T) {
+	server := newPluginHostServiceServer("local.example.plugin", nil, nil)
+	_, err := server.KVGet(context.Background(), &pluginv1.KVGetRequest{Namespace: "state", Key: "k"})
+	require.Error(t, err)
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+
+	// pluginKey 为空同样视为不可用，避免键前缀塌缩导致的跨插件泄漏。
+	blank := newPluginHostServiceServer("", newFakePluginKVStore(), nil)
+	_, err = blank.KVGet(context.Background(), &pluginv1.KVGetRequest{Namespace: "state", Key: "k"})
+	require.Error(t, err)
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+
+	// pluginKey 含 ':' 会破坏内部键结构，必须判定为不可用而非放行。
+	tainted := newPluginHostServiceServer("bad:key", newFakePluginKVStore(), nil)
+	_, err = tainted.KVGet(context.Background(), &pluginv1.KVGetRequest{Namespace: "state", Key: "k"})
+	require.Error(t, err)
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+}
+
+// pluginKey 的长度上限必须覆盖清单 id 的 maxLength（160），否则长 id 的合法插件会被
+// 静默剥夺宿主服务能力。
+func TestPluginHostServiceServer_PluginKeyLengthBoundary(t *testing.T) {
+	ctx := context.Background()
+	maxLen := newPluginHostServiceServer(strings.Repeat("a", pluginKVMaxPluginKeyLen), newFakePluginKVStore(), nil)
+	_, err := maxLen.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: []byte("v")})
+	require.NoError(t, err)
+
+	tooLong := newPluginHostServiceServer(strings.Repeat("a", pluginKVMaxPluginKeyLen+1), newFakePluginKVStore(), nil)
+	_, err = tooLong.KVSet(ctx, &pluginv1.KVSetRequest{Namespace: "state", Key: "k", Value: []byte("v")})
+	require.Error(t, err)
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+}
+
+type fakeAccountDirectory struct {
+	ids      []int64
+	identity *PluginOutboundIdentity
+	lastReq  int64
+}
+
+func (f *fakeAccountDirectory) ListPluginAccounts(_ context.Context, _, _ string) ([]int64, error) {
+	return f.ids, nil
+}
+
+func (f *fakeAccountDirectory) ResolvePluginOutboundIdentity(_ context.Context, accountID int64) (*PluginOutboundIdentity, error) {
+	f.lastReq = accountID
+	if f.identity == nil || f.identity.AccountID != accountID {
+		return nil, nil
+	}
+	return f.identity, nil
+}
+
+func TestPluginHostServiceServer_AccountDirectory(t *testing.T) {
+	ctx := context.Background()
+	dir := &fakeAccountDirectory{
+		ids: []int64{3, 7},
+		identity: &PluginOutboundIdentity{
+			AccountID: 7, Platform: "openai", AccountType: "oauth", ProxyURL: "http://p:1",
+			Token: "tok", Headers: http.Header{"Originator": {"codex-tui"}},
+		},
+	}
+	server := newPluginHostServiceServer("local.example.plugin", newFakePluginKVStore(), dir)
+
+	list, err := server.ListAccounts(ctx, &pluginv1.ListAccountsRequest{Platform: "openai", AccountType: "oauth"})
+	require.NoError(t, err)
+	assert.Equal(t, []int64{3, 7}, list.AccountIds)
+
+	resolved, err := server.ResolveOutboundIdentity(ctx, &pluginv1.ResolveOutboundIdentityRequest{AccountId: 7})
+	require.NoError(t, err)
+	require.True(t, resolved.Found)
+	assert.Equal(t, "tok", resolved.Token)
+	assert.Equal(t, "http://p:1", resolved.ProxyUrl)
+	assert.Equal(t, "codex-tui", resolved.Headers["Originator"].Values[0])
+
+	// unknown account → found=false, no error
+	miss, err := server.ResolveOutboundIdentity(ctx, &pluginv1.ResolveOutboundIdentityRequest{AccountId: 99})
+	require.NoError(t, err)
+	assert.False(t, miss.Found)
+
+	// invalid account id → InvalidArgument
+	_, err = server.ResolveOutboundIdentity(ctx, &pluginv1.ResolveOutboundIdentityRequest{AccountId: 0})
+	assert.Equal(t, codes.InvalidArgument, status.Code(err))
+}
+
+// 无目录时账号目录 RPC 必须返回 Unavailable（KV 仍可用），保证未授权插件拿不到凭据。
+func TestPluginHostServiceServer_DirectoryUnavailableWithoutDirectory(t *testing.T) {
+	server := newPluginHostServiceServer("local.example.plugin", newFakePluginKVStore(), nil)
+	_, err := server.ListAccounts(context.Background(), &pluginv1.ListAccountsRequest{})
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+	_, err = server.ResolveOutboundIdentity(context.Background(), &pluginv1.ResolveOutboundIdentityRequest{AccountId: 1})
+	assert.Equal(t, codes.Unavailable, status.Code(err))
+	// KV 仍可用
+	_, kvErr := server.KVGet(context.Background(), &pluginv1.KVGetRequest{Namespace: "state", Key: "k"})
+	require.NoError(t, kvErr)
+}
+
+func TestPluginDeclaresOpenAIOAuthCapability(t *testing.T) {
+	match := PluginManifest{Capabilities: []PluginCapability{{ID: PluginCapabilityOpenAIOAuthOutbound, Platform: PlatformOpenAI, AccountType: AccountTypeOAuth}}}
+	if !pluginDeclaresOpenAIOAuthCapability(match) {
+		t.Fatal("matching capability must be recognized")
+	}
+	wrongType := PluginManifest{Capabilities: []PluginCapability{{ID: PluginCapabilityOpenAIOAuthOutbound, Platform: PlatformOpenAI, AccountType: "apikey"}}}
+	if pluginDeclaresOpenAIOAuthCapability(wrongType) {
+		t.Fatal("wrong account_type must not match")
+	}
+	empty := PluginManifest{}
+	if pluginDeclaresOpenAIOAuthCapability(empty) {
+		t.Fatal("no capability must not match")
+	}
+}
+
+// buildHostServices 只对声明了 OpenAI OAuth 能力的插件注入账号目录。
+func TestBuildHostServicesGatesDirectoryByCapability(t *testing.T) {
+	dir := &fakeAccountDirectory{}
+	m := &PluginManager{kvStore: newFakePluginKVStore(), accountDirectory: dir}
+
+	authorized := &PluginInstallation{PluginKey: "p.authorized", Manifest: PluginManifest{
+		Capabilities: []PluginCapability{{ID: PluginCapabilityOpenAIOAuthOutbound, Platform: PlatformOpenAI, AccountType: AccountTypeOAuth}},
+	}}
+	srv, ok := m.buildHostServices(authorized).(*pluginHostServiceServer)
+	require.True(t, ok)
+	require.NotNil(t, srv.directory)
+
+	unauthorized := &PluginInstallation{PluginKey: "p.other", Manifest: PluginManifest{
+		Capabilities: []PluginCapability{{ID: "some.other.capability", Platform: "x", AccountType: "y"}},
+	}}
+	srv2, ok := m.buildHostServices(unauthorized).(*pluginHostServiceServer)
+	require.True(t, ok)
+	require.Nil(t, srv2.directory, "unauthorized plugin must not receive the account directory")
+	require.NotNil(t, srv2.store, "but KV remains available")
+}
@@ -46,6 +46,11 @@ type PluginManager struct {
 	cfg       *config.Config
 	hostInfo  PluginHostInfo
 	installer *PluginPackageInstaller
+	// kvStore 为运行中的插件提供通用宿主键值存储；为 nil 时不向插件暴露宿主服务。
+	kvStore PluginKVStore
+	// accountDirectory 为声明了对应能力的插件提供账号目录与出站身份解析（敏感能力）；
+	// 通过 SetAccountDirectory 在启动装配阶段注入，为 nil 时插件拿不到该能力。
+	accountDirectory PluginAccountDirectory

 	operationMu        sync.Mutex
 	mu                 sync.Mutex
@@ -57,13 +62,14 @@ type PluginManager struct {
 	route              atomic.Pointer[pluginRoute]
 }

-func NewPluginManager(repo PluginRepository, encryptor SecretEncryptor, cfg *config.Config, hostInfo PluginHostInfo) *PluginManager {
+func NewPluginManager(repo PluginRepository, encryptor SecretEncryptor, cfg *config.Config, hostInfo PluginHostInfo, kvStore PluginKVStore) *PluginManager {
 	return &PluginManager{
 		repo:               repo,
 		encryptor:          encryptor,
 		cfg:                cfg,
 		hostInfo:           hostInfo,
 		installer:          NewPluginPackageInstaller(cfg, hostInfo),
+		kvStore:            kvStore,
 		runtimes:           make(map[int64]*pluginRuntime),
 		localInstallations: make(map[int64]*PluginInstallation),
 	}
@@ -813,6 +819,26 @@ func (m *PluginManager) Test(ctx context.Context, id int64) (*pluginv1.TestConfi
 	return runtime.api.TestConfig(testCtx, &pluginv1.TestConfigRequest{ConfigJson: configJSON})
 }

+// Status returns the plugin's passive runtime status (its Health response, including
+// any status_json blob) for the config UI. Unlike Test it never applies config,
+// starts a temporary runtime, or reaches upstream — a plugin that is not currently
+// running simply reports "not running" with no status blob. This makes it safe to
+// serve from a lightweight, ungated, read-only endpoint used for status polling.
+func (m *PluginManager) Status(ctx context.Context, id int64) (*pluginv1.HealthResponse, error) {
+	if _, err := m.repo.GetByID(ctx, id); err != nil {
+		return nil, err
+	}
+	m.mu.Lock()
+	runtime := m.runtimes[id]
+	m.mu.Unlock()
+	if runtime == nil {
+		return &pluginv1.HealthResponse{Healthy: false, Message: "插件未运行"}, nil
+	}
+	statusCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	return runtime.status(statusCtx)
+}
+
 type pluginUIAssetClaims struct {
 	Version  int   `json:"version"`
 	PluginID int64 `json:"plugin_id"`
@@ -1018,7 +1044,44 @@ func (m *PluginManager) newRuntime(ctx context.Context, installation *PluginInst
 		return nil, err
 	}
 	timeout := time.Duration(m.cfg.Plugins.StartTimeoutSeconds) * time.Second
-	return startPluginRuntime(ctx, installation, timeout, socketDir)
+	return startPluginRuntime(ctx, installation, timeout, socketDir, m.buildHostServices(installation))
+}
+
+// SetAccountDirectory 注入账号目录实现（敏感能力）。仅在启动装配阶段调用一次，
+// 早于 Start，因此运行期读取无需额外同步。
+func (m *PluginManager) SetAccountDirectory(directory PluginAccountDirectory) {
+	m.mu.Lock()
+	m.accountDirectory = directory
+	m.mu.Unlock()
+}
+
+// buildHostServices 为单个插件构造绑定其 pluginKey 的宿主服务端点。返回 nil（未配置
+// 键值存储或缺少 pluginKey）时，startPluginRuntime 不会向插件暴露任何宿主服务。
+// 账号目录（会向插件交付账号凭据）只对「清单声明了 OpenAI OAuth 出站能力」的插件开放，
+// 从而把凭据暴露面收敛到本就要处理这些账号的插件。
+func (m *PluginManager) buildHostServices(installation *PluginInstallation) pluginv1.HostServiceServer {
+	if m.kvStore == nil || installation == nil || strings.TrimSpace(installation.PluginKey) == "" {
+		return nil
+	}
+	var directory PluginAccountDirectory
+	if pluginDeclaresOpenAIOAuthCapability(installation.Manifest) {
+		m.mu.Lock()
+		directory = m.accountDirectory
+		m.mu.Unlock()
+	}
+	return newPluginHostServiceServer(installation.PluginKey, m.kvStore, directory)
+}
+
+// pluginDeclaresOpenAIOAuthCapability reports whether the (install-validated)
+// manifest declares the OpenAI OAuth outbound transport capability.
+func pluginDeclaresOpenAIOAuthCapability(manifest PluginManifest) bool {
+	for _, capability := range manifest.Capabilities {
+		if capability.ID == PluginCapabilityOpenAIOAuthOutbound &&
+			capability.Platform == PlatformOpenAI && capability.AccountType == AccountTypeOAuth {
+			return true
+		}
+	}
+	return false
 }

 func (m *PluginManager) removeRuntimeLocked(id int64) *pluginRuntime {
@@ -112,3 +112,24 @@ func TestStablePluginBucketIsDeterministicAndBounded(t *testing.T) {
 		assert.Less(t, first, uint64(100))
 	}
 }
+
+type statusStubRepository struct {
+	PluginRepository
+}
+
+func (r *statusStubRepository) GetByID(context.Context, int64) (*PluginInstallation, error) {
+	return &PluginInstallation{ID: 7}, nil
+}
+
+// Status is the read-only, ungated runtime-status channel: when the plugin is not
+// running it must report "not running" with no status blob and never error or start
+// a runtime (that side-effecting behaviour belongs to Test, not Status).
+func TestPluginManagerStatusReportsNotRunningWithoutRuntime(t *testing.T) {
+	manager := &PluginManager{repo: &statusStubRepository{}}
+	resp, err := manager.Status(context.Background(), 7)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.False(t, resp.Healthy)
+	assert.Equal(t, "插件未运行", resp.Message)
+	assert.Empty(t, resp.StatusJson)
+}
@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"net/http"
 	"os/exec"
 	"strconv"
@@ -19,6 +20,9 @@ import (
 	pluginv1 "github.com/Wei-Shaw/sub2api/pkg/pluginapi/v1"
 	hclog "github.com/hashicorp/go-hclog"
 	hcplugin "github.com/hashicorp/go-plugin"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )

 type pluginRuntime struct {
@@ -31,7 +35,7 @@ type pluginRuntime struct {
 	doneOnce     sync.Once
 }

-func startPluginRuntime(ctx context.Context, installation *PluginInstallation, startTimeout time.Duration, socketDir string) (*pluginRuntime, error) {
+func startPluginRuntime(ctx context.Context, installation *PluginInstallation, startTimeout time.Duration, socketDir string, hostServices pluginv1.HostServiceServer) (*pluginRuntime, error) {
 	if installation == nil {
 		return nil, errors.New("插件安装记录为空")
 	}
@@ -66,11 +70,12 @@ func startPluginRuntime(ctx context.Context, installation *PluginInstallation, s
 		client.Kill()
 		return nil, fmt.Errorf("获取插件传输能力: %w", err)
 	}
-	api, ok := dispensed.(pluginv1.TransportPluginClient)
-	if !ok {
+	transportClient, ok := dispensed.(*pluginv1.TransportClient)
+	if !ok || transportClient.TransportPluginClient == nil {
 		client.Kill()
 		return nil, errors.New("插件未实现传输 gRPC 客户端")
 	}
+	api := transportClient.TransportPluginClient
 	runtime := &pluginRuntime{
 		installation: installation,
 		client:       client,
@@ -97,9 +102,56 @@ func startPluginRuntime(ctx context.Context, installation *PluginInstallation, s
 		}
 		return nil, fmt.Errorf("插件不健康: %s", health.Message)
 	}
+	// 可选地把宿主服务（HostService）反向暴露给插件。这是叠加在传输契约之上的能力：
+	// 老插件不实现 InitHostServices（返回 Unimplemented），此处静默跳过，绝不阻断启动。
+	offerPluginHostServices(ctx, installation, api, transportClient.Broker, hostServices, startTimeout)
 	return runtime, nil
 }

+// offerPluginHostServices 在 go-plugin broker 上启动一个宿主服务实例，并通过
+// InitHostServices 把 broker 流 id 交给插件。服务生命周期与插件进程绑定：client.Kill()
+// 会关闭 broker，AcceptAndServe 随之 GracefulStop，无需手动清理。整个过程尽力而为，
+// 任何失败都只记录日志、不影响插件转发能力。
+func offerPluginHostServices(
+	ctx context.Context,
+	installation *PluginInstallation,
+	api pluginv1.TransportPluginClient,
+	broker *hcplugin.GRPCBroker,
+	hostServices pluginv1.HostServiceServer,
+	startTimeout time.Duration,
+) {
+	if broker == nil || hostServices == nil || api == nil {
+		return
+	}
+	brokerID := broker.NextId()
+	go broker.AcceptAndServe(brokerID, func(opts []grpc.ServerOption) *grpc.Server {
+		server := grpc.NewServer(opts...)
+		pluginv1.RegisterHostServiceServer(server, hostServices)
+		return server
+	})
+	initCtx, cancel := context.WithTimeout(ctx, startTimeout)
+	defer cancel()
+	resp, err := api.InitHostServices(initCtx, &pluginv1.InitHostServicesRequest{
+		HostServiceId:         brokerID,
+		HostServiceApiVersion: pluginv1.HostServiceAPIVersion,
+	})
+	pluginKey := ""
+	if installation != nil {
+		pluginKey = installation.PluginKey
+	}
+	if err != nil {
+		if status.Code(err) == codes.Unimplemented {
+			slog.Debug("plugin_host_services_unimplemented", "plugin", pluginKey)
+		} else {
+			slog.Warn("plugin_host_services_init_failed", "plugin", pluginKey, "error", err)
+		}
+		return
+	}
+	if resp != nil && !resp.Ready {
+		slog.Debug("plugin_host_services_declined", "plugin", pluginKey, "message", resp.Message)
+	}
+}
+
 func (r *pluginRuntime) validateAndApplyConfig(ctx context.Context, configJSON []byte) error {
 	_, err := r.validateAndApplyNormalizedConfig(ctx, configJSON)
 	return err
@@ -161,6 +213,23 @@ func (r *pluginRuntime) checkHealth(ctx context.Context) error {
 	return nil
 }

+// status returns the plugin's passive Health response, including any status_json
+// blob it exposes for the config UI. It performs no config apply and no upstream
+// call, so it is safe to serve from a lightweight, ungated status endpoint.
+func (r *pluginRuntime) status(ctx context.Context) (*pluginv1.HealthResponse, error) {
+	if r == nil || r.api == nil || r.client == nil || r.client.Exited() {
+		return nil, errors.New("插件进程已退出")
+	}
+	health, err := r.api.Health(ctx, &pluginv1.HealthRequest{})
+	if err != nil {
+		return nil, fmt.Errorf("插件状态查询失败: %w", err)
+	}
+	if health == nil {
+		return nil, errors.New("插件未返回状态")
+	}
+	return health, nil
+}
+
 func (r *pluginRuntime) beginRequest() bool {
 	if r == nil || r.draining.Load() {
 		return false
@@ -47,7 +47,7 @@ func TestPluginRuntimeIntegration(t *testing.T) {
 		assert.FileExists(t, filepath.Join(installation.InstallPath, filepath.FromSlash(relative)))
 	}

-	runtime, err := startPluginRuntime(context.Background(), installation, 10*time.Second, filepath.Join(root, "runtime"))
+	runtime, err := startPluginRuntime(context.Background(), installation, 10*time.Second, filepath.Join(root, "runtime"), nil)
 	require.NoError(t, err)
 	defer runtime.kill()
 	require.NoError(t, runtime.validateAndApplyConfig(context.Background(), []byte(`{
@@ -30,6 +30,21 @@
 - Sub2API 继续负责响应状态处理、SSE 解析、错误映射、用量统计、计费和下游输出。
 - 灰度比例以账号 ID 稳定分桶，未命中的 OAuth 账号继续使用原有内置路径。

+## 宿主服务（HostService）
+
+`HostService` 是宿主经 go-plugin broker 反向暴露给插件的通用能力层，独立于具体插件类型，供有状态插件使用。它通过 `TransportPlugin.InitHostServices` 在运行时协商：宿主启动后把一个 broker 流 id 交给插件，插件用它拨号回宿主并获得 `HostServiceClient`。
+
+- **可选且向后兼容**：未实现 `InitHostServices` 的旧插件返回 `Unimplemented`，宿主静默跳过，转发能力不受影响。宿主服务有独立的 `HostServiceAPIVersion`，新增能力不会改变传输契约版本，也不会使既有插件失效。
+- **随进程回收**：宿主服务实例与插件进程生命周期绑定，插件退出时 broker 关闭并自动 `GracefulStop`，无需插件手动清理。
+
+当前提供的通用设施：
+
+- **命名空间键值存储（KV）**：`KVGet` / `KVSet` / `KVDelete` / `KVList`，为插件持久化跨请求、跨副本、跨重启的状态。存储由 Redis 支撑，因此多实例部署天然共享同一份状态。
+  - **命名空间隔离**：宿主根据服务该连接的运行时注入插件身份（`pluginKey`），插件无法伪造，也无法读写其它插件的命名空间。
+  - **护栏**：`namespace` / `key` 仅允许 `[A-Za-z0-9._-]`；单值上限 256 KiB；`ttl_seconds` 为 0 表示不过期、正值有上限；`KVList` 返回条数有上限。
+
+新增宿主设施时，在 `HostService` 上追加 RPC 即可，无需改动传输契约或清单格式。
+
 ## 包结构

 ```text
@@ -61,15 +76,21 @@ ui/assets/...

 插件 UI 由包内静态文件实现，宿主使用只有 `allow-scripts` 权限的 sandbox iframe 加载。iframe 没有管理员 Token，也不能直接访问管理 API。宿主为每次打开配置页生成短时资源 URL 和独立 Bridge Token，并且同时校验消息来源窗口与 Token。

-UI 可以发送以下消息：
+UI 可以发送以下消息。消息按语义分层，鉴权与副作用一致对应（读操作免二次验证，写/主动测试需要二次验证），任何插件都可复用，不针对具体插件定制：

- `config.load`
- `config.save`
- `config.test`
- `ui.resize`
- `ui.notify`
+| 消息 | 语义 | 二次验证 | 映射的插件 RPC |
+| --- | --- | --- | --- |
+| `config.load` | 读取已保存配置 | 否 | 宿主数据库 |
+| `config.save` | 写入配置 | 是 | `ValidateConfig` + `ApplyConfig` |
+| `config.test` | 主动测试配置/连通性（可产生副作用） | 是 | `TestConfig` |
+| `plugin.status` | 读取运行时状态（无副作用） | 否 | `Health`（`status_json`） |
+| `ui.resize` / `ui.notify` | 仅 UI 交互 | — | — |

-每个请求消息带 `request_id`，宿主以 `<type>.result` 返回结果。配置整体使用 Sub2API 的密钥加密后存入数据库；运行中插件会先验证并应用新配置，数据库写入失败时恢复旧配置。
+每个请求消息带 `request_id`，宿主以 `<type>.result` 返回结果。
+
+- 配置整体使用 Sub2API 的密钥加密后存入数据库；运行中插件会先验证并应用新配置，数据库写入失败时恢复旧配置。
+- `config.test` 的结果由插件 UI 自行展示（内联或经 `ui.notify`），宿主不再对成功结果强制弹出提示，避免插件把它当作轻量状态轮询时刷屏。
+- `plugin.status` 是通用的**只读**状态通道：宿主经 `GET /admin/plugins/:id/status` 调用运行中插件的 `Health`，返回 `{healthy, message, status_json}`。`status_json` 是插件自定义的**不透明** JSON 快照（宿主不解析、不参与健康判定），插件必须以无副作用方式生成（不得应用配置、访问上游或触发探测），因此该端点只读、免二次验证。插件未运行时返回 `healthy=false` 且不含 `status_json`。这样带状态面板的插件无需滥用 `config.test` 即可展示实时状态。

 ## 协议源码

@@ -44,10 +44,13 @@ UI 到宿主：
 | `config.load` | 无 | `config` |
 | `config.save` | `config` 对象 | 规范化后的 `config` |
 | `config.test` | 无 | `result` |
+| `plugin.status` | 无 | `result`（`Health`：`{healthy, message, status_json}`） |
 | `ui.resize` | `height` | 无响应 |
 | `ui.notify` | `level`、`message` | 无响应 |

-`config.test` 在 v1 中测试已保存配置。UI 若要测试当前表单，应先调用 `config.save`。
+`config.test` 在 v1 中测试已保存配置（需二次验证，可产生副作用）。UI 若要测试当前表单，应先调用 `config.save`。
+
+`plugin.status` 是只读运行时状态通道：无副作用、免二次验证、不弹宿主提示，供状态面板轮询。它映射到插件 `Health`，`result.status_json` 是插件自定义的不透明 JSON 快照。带状态展示的插件应使用它，而不是把 `config.test` 当作状态轮询。

 ## 必须执行的校验

@@ -11,6 +11,28 @@ service TransportPlugin {
  rpc ApplyConfig(ApplyConfigRequest) returns (ApplyConfigResponse);
  rpc TestConfig(TestConfigRequest) returns (TestConfigResponse);
  rpc Forward(stream ForwardRequest) returns (stream ForwardResponse);
+  // InitHostServices hands the plugin a go-plugin broker stream id it can dial
+  // back to reach the host-provided HostService. It is an optional, generic
+  // lifecycle hook: plugins built against an older contract leave it
+  // Unimplemented and the host simply runs without offering host services.
+  rpc InitHostServices(InitHostServicesRequest) returns (InitHostServicesResponse);
+}
+
+// HostService is implemented by the host and dialed by the plugin over the
+// go-plugin broker id delivered via InitHostServices. It exposes generic,
+// capability-neutral facilities that any plugin may use; new facilities are
+// added here as new RPCs without changing the transport contract version.
+service HostService {
+  rpc KVGet(KVGetRequest) returns (KVGetResponse);
+  rpc KVSet(KVSetRequest) returns (KVSetResponse);
+  rpc KVDelete(KVDeleteRequest) returns (KVDeleteResponse);
+  rpc KVList(KVListRequest) returns (KVListResponse);
+  // Account directory: enumerate the accounts the plugin is bound to act on and
+  // resolve the outbound identity (credentials/headers/proxy) the host would
+  // attach to a live request. The host restricts both calls to accounts the
+  // requesting plugin's capability already covers.
+  rpc ListAccounts(ListAccountsRequest) returns (ListAccountsResponse);
+  rpc ResolveOutboundIdentity(ResolveOutboundIdentityRequest) returns (ResolveOutboundIdentityResponse);
 }

 message GetInfoRequest {}
@@ -28,6 +50,12 @@ message HealthRequest {}
 message HealthResponse {
  bool healthy = 1;
  string message = 2;
+  // status_json is an optional plugin-defined JSON blob describing current runtime
+  // state (e.g. a status summary for the config UI). It MUST be produced passively
+  // — reading it must not apply config, reach upstream, or cause side effects — so
+  // the host can expose it via a lightweight, ungated status endpoint. Ignored by
+  // the host's health gating.
+  string status_json = 3;
 }

 message ValidateConfigRequest {
@@ -57,6 +85,9 @@ message TestConfigResponse {
  bool success = 1;
  string message = 2;
  int64 latency_ms = 3;
+  // status_json is an optional plugin-defined JSON blob surfaced to the config UI
+  // alongside the test result (e.g. a runtime status summary). Ignored by the host.
+  string status_json = 4;
 }

 message HeaderValues {
@@ -115,3 +146,79 @@ message ForwardResponse {
    ForwardResponseError error = 4;
  }
 }
+
+message InitHostServicesRequest {
+  // host_service_id is the broker stream id the plugin dials to reach HostService.
+  uint32 host_service_id = 1;
+  // host_service_api_version lets the plugin reject a host it cannot talk to.
+  uint32 host_service_api_version = 2;
+}
+
+message InitHostServicesResponse {
+  // ready is true once the plugin has successfully dialed back the host service.
+  bool ready = 1;
+  string message = 2;
+}
+
+// KV messages back a namespaced, per-plugin persistent key-value store. The
+// host derives the owning plugin from the runtime that serves the connection,
+// so a plugin can never read or write another plugin's namespace.
+message KVGetRequest {
+  string namespace = 1;
+  string key = 2;
+}
+
+message KVGetResponse {
+  bool found = 1;
+  bytes value = 2;
+}
+
+message KVSetRequest {
+  string namespace = 1;
+  string key = 2;
+  bytes value = 3;
+  // ttl_seconds > 0 sets an expiry; 0 stores the entry without expiry.
+  int64 ttl_seconds = 4;
+}
+
+message KVSetResponse {}
+
+message KVDeleteRequest {
+  string namespace = 1;
+  string key = 2;
+}
+
+message KVDeleteResponse {}
+
+message KVListRequest {
+  string namespace = 1;
+  string key_prefix = 2;
+  int32 limit = 3;
+}
+
+message KVListResponse {
+  repeated string keys = 1;
+}
+
+message ListAccountsRequest {
+  string platform = 1;
+  string account_type = 2;
+}
+
+message ListAccountsResponse {
+  repeated int64 account_ids = 1;
+}
+
+message ResolveOutboundIdentityRequest {
+  int64 account_id = 1;
+}
+
+message ResolveOutboundIdentityResponse {
+  bool found = 1;
+  int64 account_id = 2;
+  string platform = 3;
+  string account_type = 4;
+  string proxy_url = 5;
+  string token = 6;
+  map<string, HeaderValues> headers = 7;
+}
@@ -19,12 +19,13 @@ import (
 const _ = grpc.SupportPackageIsVersion9

 const (
-	TransportPlugin_GetInfo_FullMethodName        = "/sub2api.plugin.v1.TransportPlugin/GetInfo"
-	TransportPlugin_Health_FullMethodName         = "/sub2api.plugin.v1.TransportPlugin/Health"
-	TransportPlugin_ValidateConfig_FullMethodName = "/sub2api.plugin.v1.TransportPlugin/ValidateConfig"
-	TransportPlugin_ApplyConfig_FullMethodName    = "/sub2api.plugin.v1.TransportPlugin/ApplyConfig"
-	TransportPlugin_TestConfig_FullMethodName     = "/sub2api.plugin.v1.TransportPlugin/TestConfig"
-	TransportPlugin_Forward_FullMethodName        = "/sub2api.plugin.v1.TransportPlugin/Forward"
+	TransportPlugin_GetInfo_FullMethodName          = "/sub2api.plugin.v1.TransportPlugin/GetInfo"
+	TransportPlugin_Health_FullMethodName           = "/sub2api.plugin.v1.TransportPlugin/Health"
+	TransportPlugin_ValidateConfig_FullMethodName   = "/sub2api.plugin.v1.TransportPlugin/ValidateConfig"
+	TransportPlugin_ApplyConfig_FullMethodName      = "/sub2api.plugin.v1.TransportPlugin/ApplyConfig"
+	TransportPlugin_TestConfig_FullMethodName       = "/sub2api.plugin.v1.TransportPlugin/TestConfig"
+	TransportPlugin_Forward_FullMethodName          = "/sub2api.plugin.v1.TransportPlugin/Forward"
+	TransportPlugin_InitHostServices_FullMethodName = "/sub2api.plugin.v1.TransportPlugin/InitHostServices"
 )

 // TransportPluginClient is the client API for TransportPlugin service.
@@ -37,6 +38,11 @@ type TransportPluginClient interface {
 	ApplyConfig(ctx context.Context, in *ApplyConfigRequest, opts ...grpc.CallOption) (*ApplyConfigResponse, error)
 	TestConfig(ctx context.Context, in *TestConfigRequest, opts ...grpc.CallOption) (*TestConfigResponse, error)
 	Forward(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[ForwardRequest, ForwardResponse], error)
+	// InitHostServices hands the plugin a go-plugin broker stream id it can dial
+	// back to reach the host-provided HostService. It is an optional, generic
+	// lifecycle hook: plugins built against an older contract leave it
+	// Unimplemented and the host simply runs without offering host services.
+	InitHostServices(ctx context.Context, in *InitHostServicesRequest, opts ...grpc.CallOption) (*InitHostServicesResponse, error)
 }

 type transportPluginClient struct {
@@ -110,6 +116,16 @@ func (c *transportPluginClient) Forward(ctx context.Context, opts ...grpc.CallOp
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type TransportPlugin_ForwardClient = grpc.BidiStreamingClient[ForwardRequest, ForwardResponse]

+func (c *transportPluginClient) InitHostServices(ctx context.Context, in *InitHostServicesRequest, opts ...grpc.CallOption) (*InitHostServicesResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(InitHostServicesResponse)
+	err := c.cc.Invoke(ctx, TransportPlugin_InitHostServices_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // TransportPluginServer is the server API for TransportPlugin service.
 // All implementations must embed UnimplementedTransportPluginServer
 // for forward compatibility.
@@ -120,6 +136,11 @@ type TransportPluginServer interface {
 	ApplyConfig(context.Context, *ApplyConfigRequest) (*ApplyConfigResponse, error)
 	TestConfig(context.Context, *TestConfigRequest) (*TestConfigResponse, error)
 	Forward(grpc.BidiStreamingServer[ForwardRequest, ForwardResponse]) error
+	// InitHostServices hands the plugin a go-plugin broker stream id it can dial
+	// back to reach the host-provided HostService. It is an optional, generic
+	// lifecycle hook: plugins built against an older contract leave it
+	// Unimplemented and the host simply runs without offering host services.
+	InitHostServices(context.Context, *InitHostServicesRequest) (*InitHostServicesResponse, error)
 	mustEmbedUnimplementedTransportPluginServer()
 }

@@ -148,6 +169,9 @@ func (UnimplementedTransportPluginServer) TestConfig(context.Context, *TestConfi
 func (UnimplementedTransportPluginServer) Forward(grpc.BidiStreamingServer[ForwardRequest, ForwardResponse]) error {
 	return status.Error(codes.Unimplemented, "method Forward not implemented")
 }
+func (UnimplementedTransportPluginServer) InitHostServices(context.Context, *InitHostServicesRequest) (*InitHostServicesResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method InitHostServices not implemented")
+}
 func (UnimplementedTransportPluginServer) mustEmbedUnimplementedTransportPluginServer() {}
 func (UnimplementedTransportPluginServer) testEmbeddedByValue()                         {}

@@ -266,6 +290,24 @@ func _TransportPlugin_Forward_Handler(srv interface{}, stream grpc.ServerStream)
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type TransportPlugin_ForwardServer = grpc.BidiStreamingServer[ForwardRequest, ForwardResponse]

+func _TransportPlugin_InitHostServices_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(InitHostServicesRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(TransportPluginServer).InitHostServices(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: TransportPlugin_InitHostServices_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(TransportPluginServer).InitHostServices(ctx, req.(*InitHostServicesRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // TransportPlugin_ServiceDesc is the grpc.ServiceDesc for TransportPlugin service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -293,6 +335,10 @@ var TransportPlugin_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "TestConfig",
 			Handler:    _TransportPlugin_TestConfig_Handler,
 		},
+		{
+			MethodName: "InitHostServices",
+			Handler:    _TransportPlugin_InitHostServices_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{
@@ -304,3 +350,313 @@ var TransportPlugin_ServiceDesc = grpc.ServiceDesc{
 	},
 	Metadata: "plugin.proto",
 }
+
+const (
+	HostService_KVGet_FullMethodName                   = "/sub2api.plugin.v1.HostService/KVGet"
+	HostService_KVSet_FullMethodName                   = "/sub2api.plugin.v1.HostService/KVSet"
+	HostService_KVDelete_FullMethodName                = "/sub2api.plugin.v1.HostService/KVDelete"
+	HostService_KVList_FullMethodName                  = "/sub2api.plugin.v1.HostService/KVList"
+	HostService_ListAccounts_FullMethodName            = "/sub2api.plugin.v1.HostService/ListAccounts"
+	HostService_ResolveOutboundIdentity_FullMethodName = "/sub2api.plugin.v1.HostService/ResolveOutboundIdentity"
+)
+
+// HostServiceClient is the client API for HostService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// HostService is implemented by the host and dialed by the plugin over the
+// go-plugin broker id delivered via InitHostServices. It exposes generic,
+// capability-neutral facilities that any plugin may use; new facilities are
+// added here as new RPCs without changing the transport contract version.
+type HostServiceClient interface {
+	KVGet(ctx context.Context, in *KVGetRequest, opts ...grpc.CallOption) (*KVGetResponse, error)
+	KVSet(ctx context.Context, in *KVSetRequest, opts ...grpc.CallOption) (*KVSetResponse, error)
+	KVDelete(ctx context.Context, in *KVDeleteRequest, opts ...grpc.CallOption) (*KVDeleteResponse, error)
+	KVList(ctx context.Context, in *KVListRequest, opts ...grpc.CallOption) (*KVListResponse, error)
+	// Account directory: enumerate the accounts the plugin is bound to act on and
+	// resolve the outbound identity (credentials/headers/proxy) the host would
+	// attach to a live request. The host restricts both calls to accounts the
+	// requesting plugin's capability already covers.
+	ListAccounts(ctx context.Context, in *ListAccountsRequest, opts ...grpc.CallOption) (*ListAccountsResponse, error)
+	ResolveOutboundIdentity(ctx context.Context, in *ResolveOutboundIdentityRequest, opts ...grpc.CallOption) (*ResolveOutboundIdentityResponse, error)
+}
+
+type hostServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewHostServiceClient(cc grpc.ClientConnInterface) HostServiceClient {
+	return &hostServiceClient{cc}
+}
+
+func (c *hostServiceClient) KVGet(ctx context.Context, in *KVGetRequest, opts ...grpc.CallOption) (*KVGetResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(KVGetResponse)
+	err := c.cc.Invoke(ctx, HostService_KVGet_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *hostServiceClient) KVSet(ctx context.Context, in *KVSetRequest, opts ...grpc.CallOption) (*KVSetResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(KVSetResponse)
+	err := c.cc.Invoke(ctx, HostService_KVSet_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *hostServiceClient) KVDelete(ctx context.Context, in *KVDeleteRequest, opts ...grpc.CallOption) (*KVDeleteResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(KVDeleteResponse)
+	err := c.cc.Invoke(ctx, HostService_KVDelete_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *hostServiceClient) KVList(ctx context.Context, in *KVListRequest, opts ...grpc.CallOption) (*KVListResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(KVListResponse)
+	err := c.cc.Invoke(ctx, HostService_KVList_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *hostServiceClient) ListAccounts(ctx context.Context, in *ListAccountsRequest, opts ...grpc.CallOption) (*ListAccountsResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ListAccountsResponse)
+	err := c.cc.Invoke(ctx, HostService_ListAccounts_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *hostServiceClient) ResolveOutboundIdentity(ctx context.Context, in *ResolveOutboundIdentityRequest, opts ...grpc.CallOption) (*ResolveOutboundIdentityResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ResolveOutboundIdentityResponse)
+	err := c.cc.Invoke(ctx, HostService_ResolveOutboundIdentity_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// HostServiceServer is the server API for HostService service.
+// All implementations must embed UnimplementedHostServiceServer
+// for forward compatibility.
+//
+// HostService is implemented by the host and dialed by the plugin over the
+// go-plugin broker id delivered via InitHostServices. It exposes generic,
+// capability-neutral facilities that any plugin may use; new facilities are
+// added here as new RPCs without changing the transport contract version.
+type HostServiceServer interface {
+	KVGet(context.Context, *KVGetRequest) (*KVGetResponse, error)
+	KVSet(context.Context, *KVSetRequest) (*KVSetResponse, error)
+	KVDelete(context.Context, *KVDeleteRequest) (*KVDeleteResponse, error)
+	KVList(context.Context, *KVListRequest) (*KVListResponse, error)
+	// Account directory: enumerate the accounts the plugin is bound to act on and
+	// resolve the outbound identity (credentials/headers/proxy) the host would
+	// attach to a live request. The host restricts both calls to accounts the
+	// requesting plugin's capability already covers.
+	ListAccounts(context.Context, *ListAccountsRequest) (*ListAccountsResponse, error)
+	ResolveOutboundIdentity(context.Context, *ResolveOutboundIdentityRequest) (*ResolveOutboundIdentityResponse, error)
+	mustEmbedUnimplementedHostServiceServer()
+}
+
+// UnimplementedHostServiceServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedHostServiceServer struct{}
+
+func (UnimplementedHostServiceServer) KVGet(context.Context, *KVGetRequest) (*KVGetResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method KVGet not implemented")
+}
+func (UnimplementedHostServiceServer) KVSet(context.Context, *KVSetRequest) (*KVSetResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method KVSet not implemented")
+}
+func (UnimplementedHostServiceServer) KVDelete(context.Context, *KVDeleteRequest) (*KVDeleteResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method KVDelete not implemented")
+}
+func (UnimplementedHostServiceServer) KVList(context.Context, *KVListRequest) (*KVListResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method KVList not implemented")
+}
+func (UnimplementedHostServiceServer) ListAccounts(context.Context, *ListAccountsRequest) (*ListAccountsResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ListAccounts not implemented")
+}
+func (UnimplementedHostServiceServer) ResolveOutboundIdentity(context.Context, *ResolveOutboundIdentityRequest) (*ResolveOutboundIdentityResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method ResolveOutboundIdentity not implemented")
+}
+func (UnimplementedHostServiceServer) mustEmbedUnimplementedHostServiceServer() {}
+func (UnimplementedHostServiceServer) testEmbeddedByValue()                     {}
+
+// UnsafeHostServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to HostServiceServer will
+// result in compilation errors.
+type UnsafeHostServiceServer interface {
+	mustEmbedUnimplementedHostServiceServer()
+}
+
+func RegisterHostServiceServer(s grpc.ServiceRegistrar, srv HostServiceServer) {
+	// If the following call panics, it indicates UnimplementedHostServiceServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&HostService_ServiceDesc, srv)
+}
+
+func _HostService_KVGet_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(KVGetRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).KVGet(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_KVGet_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).KVGet(ctx, req.(*KVGetRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _HostService_KVSet_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(KVSetRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).KVSet(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_KVSet_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).KVSet(ctx, req.(*KVSetRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _HostService_KVDelete_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(KVDeleteRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).KVDelete(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_KVDelete_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).KVDelete(ctx, req.(*KVDeleteRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _HostService_KVList_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(KVListRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).KVList(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_KVList_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).KVList(ctx, req.(*KVListRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _HostService_ListAccounts_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListAccountsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).ListAccounts(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_ListAccounts_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).ListAccounts(ctx, req.(*ListAccountsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _HostService_ResolveOutboundIdentity_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ResolveOutboundIdentityRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(HostServiceServer).ResolveOutboundIdentity(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: HostService_ResolveOutboundIdentity_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(HostServiceServer).ResolveOutboundIdentity(ctx, req.(*ResolveOutboundIdentityRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// HostService_ServiceDesc is the grpc.ServiceDesc for HostService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var HostService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "sub2api.plugin.v1.HostService",
+	HandlerType: (*HostServiceServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "KVGet",
+			Handler:    _HostService_KVGet_Handler,
+		},
+		{
+			MethodName: "KVSet",
+			Handler:    _HostService_KVSet_Handler,
+		},
+		{
+			MethodName: "KVDelete",
+			Handler:    _HostService_KVDelete_Handler,
+		},
+		{
+			MethodName: "KVList",
+			Handler:    _HostService_KVList_Handler,
+		},
+		{
+			MethodName: "ListAccounts",
+			Handler:    _HostService_ListAccounts_Handler,
+		},
+		{
+			MethodName: "ResolveOutboundIdentity",
+			Handler:    _HostService_ResolveOutboundIdentity_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "plugin.proto",
+}
@@ -14,6 +14,10 @@ const (
 	TransportAPIVersion = 1
 	// UIBridgeVersion 是插件管理页与沙箱 UI 的消息协议版本。
 	UIBridgeVersion = 1
+	// HostServiceAPIVersion 是宿主反向服务（HostService）的契约版本。它独立于
+	// TransportAPIVersion：宿主服务是叠加在传输契约之上的可选能力，通过
+	// InitHostServices 在运行时协商，因此新增宿主能力不会使既有插件失效。
+	HostServiceAPIVersion = 1
 	// TransportPluginName 是 go-plugin 中注册的唯一能力名称。
 	TransportPluginName = "oauth_transport"
 )
@@ -25,19 +29,38 @@ var HandshakeConfig = hcplugin.HandshakeConfig{
 	MagicCookieValue: "sub2api-plugin-v1",
 }

+// HostBrokerReceiver 由需要反向调用宿主服务的插件传输实现选择性实现。
+// GRPCServer 在插件进程启动注册服务时把 go-plugin broker 交给它，插件随后可在
+// InitHostServices 中用宿主下发的 broker id 拨号回宿主。宿主侧不实现此接口。
+type HostBrokerReceiver interface {
+	SetHostBroker(broker *hcplugin.GRPCBroker)
+}
+
+// TransportClient 是宿主 Dispense 传输能力后拿到的句柄，除 gRPC 客户端外还捆绑了
+// 本连接的 go-plugin broker，宿主据此把 HostService 反向暴露给插件。
+type TransportClient struct {
+	TransportPluginClient
+	Broker *hcplugin.GRPCBroker
+}
+
 // GRPCPlugin 把生成的 gRPC 服务注册到 go-plugin 子进程。
 type GRPCPlugin struct {
 	hcplugin.NetRPCUnsupportedPlugin
 	Impl TransportPluginServer
 }

-func (p *GRPCPlugin) GRPCServer(_ *hcplugin.GRPCBroker, server *grpc.Server) error {
+func (p *GRPCPlugin) GRPCServer(broker *hcplugin.GRPCBroker, server *grpc.Server) error {
 	RegisterTransportPluginServer(server, p.Impl)
+	// 插件侧：把 broker 交给需要反向访问宿主服务的实现（可选能力）。
+	if receiver, ok := p.Impl.(HostBrokerReceiver); ok {
+		receiver.SetHostBroker(broker)
+	}
 	return nil
 }

-func (p *GRPCPlugin) GRPCClient(_ context.Context, _ *hcplugin.GRPCBroker, conn *grpc.ClientConn) (any, error) {
-	return NewTransportPluginClient(conn), nil
+func (p *GRPCPlugin) GRPCClient(_ context.Context, broker *hcplugin.GRPCBroker, conn *grpc.ClientConn) (any, error) {
+	// 宿主侧：捆绑 broker，供 startPluginRuntime 反向暴露 HostService。
+	return &TransportClient{TransportPluginClient: NewTransportPluginClient(conn), Broker: broker}, nil
 }

 // ClientPluginMap 返回宿主侧使用的插件声明。
@@ -75,6 +75,13 @@ export interface PluginTestResult {
  success: boolean
  message: string
  latency_ms: number
+  status_json?: string
+}
+
+export interface PluginStatusResult {
+  healthy: boolean
+  message: string
+  status_json?: string
 }

 export interface PluginUISession {
@@ -138,6 +145,11 @@ export async function test(id: number): Promise<PluginTestResult> {
  return data
 }

+export async function status(id: number): Promise<PluginStatusResult> {
+  const { data } = await apiClient.get<PluginStatusResult>(`/admin/plugins/${id}/status`)
+  return data
+}
+
 export async function createUISession(id: number): Promise<PluginUISession> {
  const { data } = await apiClient.post<PluginUISession>(`/admin/plugins/${id}/ui-session`)
  return data
@@ -152,5 +164,6 @@ export default {
  getConfig,
  saveConfig,
  test,
+  status,
  createUISession
 }
@@ -611,7 +611,8 @@ async function handleBridgeMessage(event: MessageEvent): Promise<void> {
  const expectsResponse =
    message.type === "config.load" ||
    message.type === "config.save" ||
-    message.type === "config.test";
+    message.type === "config.test" ||
+    message.type === "plugin.status";
  if (expectsResponse) {
    if (!requestID || pendingBridgeRequests.has(requestID)) return;
    registerBridgeRequest(requestID);
@@ -650,11 +651,22 @@ async function handleBridgeMessage(event: MessageEvent): Promise<void> {
          adminAPI.plugins.test(configPlugin.value!.id),
        );
        postBridgeResult(message, { ok: result.success, result });
-        if (result.success)
-          appStore.showSuccess(
-            result.message || t("admin.plugins.testSuccess"),
-          );
-        else appStore.showError(result.message || t("common.error"));
+        // A successful result is delivered back to the plugin UI, which owns how it
+        // presents it (inline status, or an explicit ui.notify). Only force a host
+        // toast on failure so genuine errors are never silently dropped — plugins
+        // may call config.test for lightweight status polling, not just as an
+        // explicit "test" action, and those must not spam a success toast.
+        if (!result.success)
+          appStore.showError(result.message || t("common.error"));
+        break;
+      }
+      case "plugin.status": {
+        // Read-only runtime status (the plugin's Health snapshot). It has no side
+        // effects, so it is intentionally NOT step-up gated and never raises a host
+        // toast — the plugin UI renders it however it likes. This is the generic
+        // channel for any plugin to surface live state without abusing config.test.
+        const result = await adminAPI.plugins.status(configPlugin.value!.id);
+        postBridgeResult(message, { ok: true, result });
        break;
      }
      case "ui.resize": {