From 75829fa442fc7e6345a87c99b390ba5dfb935028 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 11 Oct 2026 08:26:24 +0000 Subject: [PATCH] Add the Surface store, activation guard, and EXT-17 host capability Why: Generative UI proposal U2 needs the host side of the contract: owner-isolated surface storage with revision CAS, an execution-level enable/disable ledger (default-off, epoch bumped on every transition so in-flight submits after a disable are rejected), a single-choke-point service that validates specs (strict decode + catalog rules) before saving, event publish with pending recovery, the optional SurfaceHost extension capability, and direct API endpoints that share the same guard (no bypass by entrance). Evidence: go build/vet clean; full go test ./... passes (SQLite); surface store tests also pass under AGENT_PG_TEST_DSN (PostgreSQL); surface.txt regenerated via the AST snapshot test. Impact: Additive migration v22 (new tables only); existing endpoints and plugin_grants semantics untouched; old consumers unaffected. U3+ (data binding, actions, frontend) deliberately out of scope. Refs: Documents proposal 2026-10-10-generative-ui-plugin.md (r2 U2); Core PR #25; Dev task feature/generative-ui-u2 --- deps/agent-system-core | 2 +- internal/app/app.go | 10 +- internal/app/hostcaps.go | 56 ++++- internal/server/platform_api.go | 102 +++++++++ internal/server/server.go | 16 +- internal/server/surface_api_test.go | 89 ++++++++ internal/store/migrate.go | 80 +++++++ internal/store/surfaces.go | 297 +++++++++++++++++++++++++ internal/store/surfaces_test.go | 126 +++++++++++ internal/surfacesvc/surfacesvc.go | 202 +++++++++++++++++ internal/surfacesvc/surfacesvc_test.go | 148 ++++++++++++ pkg/extension/extension.go | 35 ++- pkg/extension/surface.txt | 6 + 13 files changed, 1157 insertions(+), 12 deletions(-) create mode 100644 internal/server/surface_api_test.go create mode 100644 internal/store/surfaces.go create mode 100644 internal/store/surfaces_test.go create mode 100644 internal/surfacesvc/surfacesvc.go create mode 100644 internal/surfacesvc/surfacesvc_test.go diff --git a/deps/agent-system-core b/deps/agent-system-core index a253a40..452daf3 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit a253a40b8e5563119b00c4e71fb61fe0d11d6d6a +Subproject commit 452daf3ffa1d94d54ccd0d84a46070ce501bc14b diff --git a/internal/app/app.go b/internal/app/app.go index 6a8b8e4..f1db8b5 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -50,6 +50,7 @@ import ( "agent-system-backend/internal/skills" "agent-system-backend/internal/squadadapter" "agent-system-backend/internal/store" + "agent-system-backend/internal/surfacesvc" "agent-system-backend/internal/tools" "agent-system-backend/internal/workflow" core "agent-system-core" @@ -472,6 +473,12 @@ func Run(opt Options) error { // 收件箱任务卡：事件走会话事件管道（持久化/seq/重放）；关闭即 T0 基线。 cardSvc := cardsvc.New(st, mgr.EmitCardEvent) + surfaceSvc := surfacesvc.New(st, mgr.EmitCardEvent) + go func() { // 启动恢复：重发「快照已落盘、事件未发出」的待恢复投影（KTD）。 + if n, err := surfaceSvc.RecoverPendingEvents(context.Background()); err == nil && n > 0 { + log.Printf("surface: recovered %d pending events", n) + } + }() mgr.SetCards(cardSvc) // 执行预算（协作存储 U3，R8）：runtime.execBudgetGlobal / runtime.execBudgetPerUser // 全零=不限（本地档默认，行为不变）；开启后执行许可共享账本生效， @@ -490,7 +497,7 @@ func Run(opt Options) error { pluginHost.OnCommands = bus.RegisterPluginCommands // hostcaps：hostapi 闭包与进程内插件句柄共用的 owner 作用域能力（EXT-1 批注 1）。 authzSlot := &projectAuthz{} - caps := hostcaps{st: st, pipe: pipe, set: setSvc, authz: authzSlot} + caps := hostcaps{st: st, pipe: pipe, set: setSvc, authz: authzSlot, surfaces: surfaceSvc} // EXT-13：回合内检索（knowledge_search / @knowledge 注入）与 HTTP/Host 检索面 // 同一适配器授权；槽为共享指针，扩展在 applyCompiled 期间注册后即生效。 mgr.SetSharedKnowledgeAllow(func(owner string, projectID core.ID) bool { @@ -555,6 +562,7 @@ func Run(opt Options) error { srv.Workflows = wfReg mgr.SetWorkflows(wfReg) srv.Cards = cardSvc + srv.Surfaces = surfaceSvc srv.Bus = bus srv.Gate = gate srv.Prompts = promptSvc diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go index bd363e8..3d0dd9c 100644 --- a/internal/app/hostcaps.go +++ b/internal/app/hostcaps.go @@ -11,6 +11,7 @@ import ( "agent-system-backend/internal/runtime" "agent-system-backend/internal/settings" "agent-system-backend/internal/store" + "agent-system-backend/internal/surfacesvc" core "agent-system-core" ) @@ -18,10 +19,11 @@ import ( // 同一实现同时服务 hostapi.Server 的 Search/Complete 闭包与进程内 // inprocHost（EXT-1 批注 1：复用现有 owner 作用域函数，不新造句柄）。 type hostcaps struct { - st *store.Store - pipe *rag.Pipeline - set *settings.Service - authz *projectAuthz // EXT-8 适配器槽（共享指针；首个注册者生效） + st *store.Store + pipe *rag.Pipeline + set *settings.Service + authz *projectAuthz // EXT-8 适配器槽（共享指针；首个注册者生效） + surfaces *surfacesvc.Service // EXT-17 Surface 能力槽（nil = 未装配，明确拒绝） } // projectAuthz EXT-8 资源授权适配器槽：进程内扩展注册，组合根装配进 Server。 @@ -230,3 +232,49 @@ func (h inprocHost) ArtifactContent(artifactID string) ([]byte, error) { func (h inprocHost) ParseDocument(ctx context.Context, filename, mime string, content []byte) (core.ParsedDocument, error) { return h.caps.parse(ctx, filename, mime, content) } + +// ---- EXT-17 Surface 能力（可选）：nil 槽=能力缺失，明确拒绝不降级 ---- + +func (h inprocHost) surface() (*surfacesvc.Service, error) { + if h.caps.surfaces == nil { + return nil, surfacesvc.ErrUnavailable + } + return h.caps.surfaces, nil +} + +func (h inprocHost) SurfaceSave(ctx context.Context, req extension.SurfaceSaveRequest) (core.SurfaceSnapshot, error) { + svc, err := h.surface() + if err != nil { + return core.SurfaceSnapshot{}, err + } + return svc.Submit(ctx, h.owner, h.pluginID, surfacesvc.SubmitRequest{ + SurfaceID: req.SurfaceID, ConversationID: req.ConversationID, SourceRunID: req.SourceRunID, + BaseRevision: req.BaseRevision, ObservedEpoch: req.ObservedEpoch, + ContextVersion: req.ContextVersion, SemanticText: req.SemanticText, Spec: req.Spec, + }) +} + +func (h inprocHost) SurfaceGet(ctx context.Context, surfaceID core.ID) (core.SurfaceSnapshot, error) { + svc, err := h.surface() + if err != nil { + return core.SurfaceSnapshot{}, err + } + return svc.Get(ctx, h.owner, surfaceID) +} + +func (h inprocHost) SurfaceList(ctx context.Context, conversationID core.ID) ([]core.SurfaceMeta, error) { + svc, err := h.surface() + if err != nil { + return nil, err + } + return svc.List(ctx, h.owner, conversationID) +} + +func (h inprocHost) SurfaceActivation(ctx context.Context) (bool, int64, error) { + svc, err := h.surface() + if err != nil { + return false, 0, err + } + row, err := svc.Activation(ctx, h.owner, h.pluginID) + return row.Enabled, row.Epoch, err +} diff --git a/internal/server/platform_api.go b/internal/server/platform_api.go index f2e8a4a..678b4ee 100644 --- a/internal/server/platform_api.go +++ b/internal/server/platform_api.go @@ -2,12 +2,14 @@ package server import ( "encoding/json" + "errors" "fmt" "net/http" "strconv" "strings" "agent-system-backend/internal/store" + "agent-system-backend/internal/surfacesvc" core "agent-system-core" ) @@ -185,3 +187,103 @@ func messageText(m core.Message) string { } return b.String() } + +// ---- Generative UI Surface（提案 U2；EXT-17）。直接 API 与插件命令共用 +// surfacesvc.Submit 单一咽喉：禁用/epoch/校验/CAS 守卫不因入口不同而绕过。---- + +func (s *Server) submitSurface(w http.ResponseWriter, r *http.Request) { + if s.Surfaces == nil { + http.Error(w, "surface capability not enabled", http.StatusNotImplemented) + return + } + var body struct { + PluginID string `json:"pluginId"` + ConversationID string `json:"conversationId"` + SurfaceID string `json:"surfaceId"` + BaseRevision int `json:"baseRevision"` + ObservedEpoch int64 `json:"observedEpoch"` + ContextVersion string `json:"contextVersion"` + SemanticText string `json:"semanticText"` + Spec json.RawMessage `json:"spec"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.PluginID == "" || len(body.Spec) == 0 { + http.Error(w, "pluginId and spec required", http.StatusBadRequest) + return + } + snap, err := s.Surfaces.Submit(r.Context(), s.owner(r), body.PluginID, surfacesvc.SubmitRequest{ + SurfaceID: core.ID(body.SurfaceID), ConversationID: core.ID(body.ConversationID), + BaseRevision: body.BaseRevision, ObservedEpoch: body.ObservedEpoch, + ContextVersion: body.ContextVersion, SemanticText: body.SemanticText, Spec: body.Spec, + }) + if err != nil { + s.writeSurfaceError(w, err) + return + } + writeJSON(w, http.StatusCreated, snap) +} + +func (s *Server) getSurface(w http.ResponseWriter, r *http.Request) { + if s.Surfaces == nil { + http.Error(w, "surface capability not enabled", http.StatusNotImplemented) + return + } + snap, err := s.Surfaces.Get(r.Context(), s.owner(r), core.ID(r.PathValue("id"))) + if err != nil { + s.writeSurfaceError(w, err) + return + } + writeJSON(w, http.StatusOK, snap) +} + +func (s *Server) listSurfaces(w http.ResponseWriter, r *http.Request) { + if s.Surfaces == nil { + http.Error(w, "surface capability not enabled", http.StatusNotImplemented) + return + } + metas, err := s.Surfaces.List(r.Context(), s.owner(r), core.ID(r.PathValue("id"))) + if err != nil { + s.writeSurfaceError(w, err) + return + } + writeJSON(w, http.StatusOK, orEmptySlice(metas)) +} + +func (s *Server) setSurfaceActivation(w http.ResponseWriter, r *http.Request) { + if s.Surfaces == nil { + http.Error(w, "surface capability not enabled", http.StatusNotImplemented) + return + } + var body struct { + Enabled bool `json:"enabled"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "enabled required", http.StatusBadRequest) + return + } + row, err := s.Surfaces.SetActivation(r.Context(), s.owner(r), r.PathValue("id"), body.Enabled) + if err != nil { + s.writeSurfaceError(w, err) + return + } + writeJSON(w, http.StatusOK, row) +} + +// writeSurfaceError 把服务层确定错误映射到状态码：禁用/epoch=409 语义（准入 +// 截止），规格违规=400 带字段清单，冲突=409 带当前版本，NotFound=404。 +func (s *Server) writeSurfaceError(w http.ResponseWriter, err error) { + var invalid *surfacesvc.ErrInvalidSpec + var conflict *store.ErrRevisionConflict + switch { + case errors.Is(err, surfacesvc.ErrDisabled), errors.Is(err, surfacesvc.ErrStaleEpoch), + errors.Is(err, surfacesvc.ErrUnavailable): + http.Error(w, err.Error(), http.StatusForbidden) + case errors.As(err, &invalid): + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid spec", "details": invalid.Errs}) + case errors.As(err, &conflict): + writeJSON(w, http.StatusConflict, map[string]any{"error": "revision conflict", "current": conflict.Current}) + case errors.Is(err, store.ErrNotFound): + http.Error(w, err.Error(), http.StatusNotFound) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 8c8dc60..5b778d3 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -31,6 +31,7 @@ import ( "agent-system-backend/internal/runmgr" "agent-system-backend/internal/settings" "agent-system-backend/internal/store" + "agent-system-backend/internal/surfacesvc" "agent-system-backend/internal/tools" "agent-system-backend/internal/workflow" "agent-system-backend/web" @@ -45,11 +46,12 @@ type Server struct { Limiter ratelimit.Limiter // 租户限流（nil = 不限；组合根注入） LimitLookup func() int // 当前限流值（次/分；<=0 不限；组合根缓存注入） mgr *runmgr.Manager - Cards *cardsvc.Service // 收件箱任务卡（nil = 不启用；组合根注入） - Bus *cmdbus.Bus // 命令总线（nil = 不启用；组合根注入；四入口同源 D27） - Gate NLIntentGate // 意图闸门（chat 入口同源判定；nil = 不接） - Workflows *workflow.Registry // 已装载工作流（nil = 不接） - Prompts *prompts.Service // 提示词库（nil = 不启用 API 与技能命令；执行器另有回退） + Cards *cardsvc.Service // 收件箱任务卡（nil = 不启用；组合根注入） + Surfaces *surfacesvc.Service // Generative UI Surface（nil = 不启用；EXT-17） + Bus *cmdbus.Bus // 命令总线（nil = 不启用；组合根注入；四入口同源 D27） + Gate NLIntentGate // 意图闸门（chat 入口同源判定；nil = 不接） + Workflows *workflow.Registry // 已装载工作流（nil = 不接） + Prompts *prompts.Service // 提示词库（nil = 不启用 API 与技能命令；执行器另有回退） reg *tools.Registry prof *profiles.Service set *settings.Service @@ -310,6 +312,10 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("DELETE /api/tools/{name}", s.deleteUserTool) mux.HandleFunc("GET /api/plugins", s.listPlugins) mux.HandleFunc("POST /api/plugins/{id}/enabled", s.setPluginEnabled) + mux.HandleFunc("POST /api/plugins/{id}/activation", s.setSurfaceActivation) + mux.HandleFunc("POST /api/surfaces", s.submitSurface) + mux.HandleFunc("GET /api/surfaces/{id}", s.getSurface) + mux.HandleFunc("GET /api/conversations/{id}/surfaces", s.listSurfaces) mux.HandleFunc("POST /api/plugins/reload", s.reloadPlugins) mux.HandleFunc("GET /api/plugins/settings-schema", s.pluginSettingsSchema) mux.HandleFunc("GET /api/registry", s.registry) diff --git a/internal/server/surface_api_test.go b/internal/server/surface_api_test.go new file mode 100644 index 0000000..c05a171 --- /dev/null +++ b/internal/server/surface_api_test.go @@ -0,0 +1,89 @@ +package server + +// Generative UI U2 红测（提案 KTD7）：直接 API 与插件命令共用 surfacesvc +// 单一咽喉——禁用状态下 POST /api/surfaces 必须被拒（不能因入口绕过执行级 +// 启禁），读取与启禁端点行为符合「历史保留 + epoch 纪元」语义。 + +import ( + "bytes" + "net/http" + "net/http/httptest" + "testing" + + "agent-system-backend/internal/surfacesvc" +) + +func TestSurfaceDirectAPIGuards(t *testing.T) { + srv, st := newAuthTestServer(t) + h := srv.Handler() + key, err := st.CreateAPIKey("alice", "e2e") + if err != nil { + t.Fatal(err) + } + svc := surfacesvc.New(st, nil) + srv.Surfaces = svc + + post := func(path, body string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body)) + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec + } + get := func(path string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, path, nil) + req.Header.Set("Authorization", "Bearer "+key) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec + } + spec := `"spec":{"title":"t","nodes":[{"type":"markdown","id":"n1","markdown":"x"}]}` + + // 未启用：提交被拒（403），提示能力存在但插件未启用 + if rec := post("/api/surfaces", `{"pluginId":"genui","conversationId":"c1",`+spec+`}`); rec.Code != http.StatusForbidden { + t.Fatalf("disabled submit should 403, got %d: %s", rec.Code, rec.Body.String()) + } + // 启用（activation 端点走服务层，epoch=1） + if rec := post("/api/plugins/genui/activation", `{"enabled":true}`); rec.Code != http.StatusOK { + t.Fatalf("enable failed: %d %s", rec.Code, rec.Body.String()) + } + // 提交成功 → 201 + rec := post("/api/surfaces", `{"pluginId":"genui","conversationId":"c1",`+spec+`}`) + if rec.Code != http.StatusCreated { + t.Fatalf("enabled submit should 201, got %d: %s", rec.Code, rec.Body.String()) + } + var snap struct { + Identity struct { + SurfaceID string `json:"surfaceId"` + } `json:"identity"` + Revision int `json:"revision"` + } + readJSON(t, rec.Body.Bytes(), &snap) + + // 读取可用；列表可用 + if rec := get("/api/surfaces/" + snap.Identity.SurfaceID); rec.Code != http.StatusOK { + t.Fatalf("get failed: %d", rec.Code) + } + if rec := get("/api/conversations/c1/surfaces"); rec.Code != http.StatusOK { + t.Fatalf("list failed: %d", rec.Code) + } + // 禁用后：提交 403；读取仍 200（历史保留） + if rec := post("/api/plugins/genui/activation", `{"enabled":false}`); rec.Code != http.StatusOK { + t.Fatalf("disable failed: %d", rec.Code) + } + if rec := post("/api/surfaces", `{"pluginId":"genui","conversationId":"c1",`+spec+`}`); rec.Code != http.StatusForbidden { + t.Fatalf("post-disable submit should 403, got %d", rec.Code) + } + if rec := get("/api/surfaces/" + snap.Identity.SurfaceID); rec.Code != http.StatusOK { + t.Fatalf("post-disable read should stay 200, got %d", rec.Code) + } + // 未带凭据：401（认证面同现有守卫） + if rec := httptest.NewRecorder(); true { + req := httptest.NewRequest(http.MethodPost, "/api/surfaces", bytes.NewBufferString(`{"pluginId":"genui"}`)) + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("unauthenticated should 401, got %d", rec.Code) + } + } +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 600daa7..18c533b 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -512,6 +512,86 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS admission_leases`, }}}, + migration{version: 22, name: "generative_surfaces", + // Generative UI U2（提案 KTD3/KTD7）：宿主通用 Surface 存储（owner 隔离 + + // revision 乐观并发 + activationEpoch 启禁纪元）。spec 存整页 JSON 快照， + // event_pending 标记「快照已落盘、投影事件未发出」的待恢复态。 + up: map[string][]string{"sqlite": { + `CREATE TABLE IF NOT EXISTS surfaces ( + surface_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + project_id TEXT NOT NULL DEFAULT '', + conversation_id TEXT NOT NULL DEFAULT '', + plugin_id TEXT NOT NULL, + compat TEXT NOT NULL DEFAULT '', + latest_revision INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_surfaces_owner_conv ON surfaces(owner, conversation_id, updated_at)`, + `CREATE TABLE IF NOT EXISTS surface_revisions ( + surface_id TEXT NOT NULL, + revision INTEGER NOT NULL, + base_revision INTEGER NOT NULL DEFAULT 0, + context_version TEXT NOT NULL DEFAULT '', + activation_epoch INTEGER NOT NULL DEFAULT 0, + spec TEXT NOT NULL, + semantic_text TEXT NOT NULL DEFAULT '', + event_pending INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + PRIMARY KEY (surface_id, revision) + )`, + `CREATE TABLE IF NOT EXISTS plugin_activation ( + owner TEXT NOT NULL, + plugin_id TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 0, + epoch INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (owner, plugin_id) + )`, + }, "postgres": { + `CREATE TABLE IF NOT EXISTS surfaces ( + surface_id TEXT PRIMARY KEY, + owner TEXT NOT NULL, + project_id TEXT NOT NULL DEFAULT '', + conversation_id TEXT NOT NULL DEFAULT '', + plugin_id TEXT NOT NULL, + compat TEXT NOT NULL DEFAULT '', + latest_revision INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_surfaces_owner_conv ON surfaces(owner, conversation_id, updated_at)`, + `CREATE TABLE IF NOT EXISTS surface_revisions ( + surface_id TEXT NOT NULL, + revision INTEGER NOT NULL, + base_revision INTEGER NOT NULL DEFAULT 0, + context_version TEXT NOT NULL DEFAULT '', + activation_epoch INTEGER NOT NULL DEFAULT 0, + spec TEXT NOT NULL, + semantic_text TEXT NOT NULL DEFAULT '', + event_pending INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + PRIMARY KEY (surface_id, revision) + )`, + `CREATE TABLE IF NOT EXISTS plugin_activation ( + owner TEXT NOT NULL, + plugin_id TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 0, + epoch INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + PRIMARY KEY (owner, plugin_id) + )`, + }}, + down: map[string][]string{"sqlite": { + `DROP TABLE IF EXISTS plugin_activation`, + `DROP TABLE IF EXISTS surface_revisions`, + `DROP TABLE IF EXISTS surfaces`, + }, "postgres": { + `DROP TABLE IF EXISTS plugin_activation`, + `DROP TABLE IF EXISTS surface_revisions`, + `DROP TABLE IF EXISTS surfaces`, + }}}, ) } diff --git a/internal/store/surfaces.go b/internal/store/surfaces.go new file mode 100644 index 0000000..f77bd94 --- /dev/null +++ b/internal/store/surfaces.go @@ -0,0 +1,297 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + core "agent-system-core" +) + +// ---- Generative UI Surface 存储（提案 U2；迁移 v21）---- + +// ErrRevisionConflict 是 revision 乐观并发冲突：携带当前最新版本供调用方对齐后重试。 +type ErrRevisionConflict struct { + Current int +} + +func (e *ErrRevisionConflict) Error() string { + return fmt.Sprintf("surface: revision conflict (current=%d)", e.Current) +} + +// SurfaceRecord 是 surfaces 表的行（页面元数据；规格在修订行）。 +type SurfaceRecord struct { + SurfaceID string + Owner string + ProjectID string + ConversationID string + PluginID string + Compat string + LatestRevision int + CreatedAt string + UpdatedAt string +} + +// SurfaceRevisionRow 是 surface_revisions 表的行。 +type SurfaceRevisionRow struct { + SurfaceID string + Revision int + BaseRevision int + ContextVersion string + ActivationEpoch int64 + Spec string + SemanticText string + EventPending bool + CreatedAt string +} + +// ActivationRow 是 plugin_activation 表的行：Surface 能力的执行级启禁状态。 +// 无行 = {Enabled:false, Epoch:0}——新插件默认关闭（KTD7，不沿用 grants 缺省 true）。 +type ActivationRow struct { + Owner string + PluginID string + Enabled bool + Epoch int64 + UpdatedAt string +} + +func (s *Store) GetActivation(owner, pluginID string) (ActivationRow, error) { + var row ActivationRow + var enabled int + err := s.db.QueryRow(s.q(`SELECT owner,plugin_id,enabled,epoch,updated_at FROM plugin_activation WHERE owner=? AND plugin_id=?`), + owner, pluginID).Scan(&row.Owner, &row.PluginID, &enabled, &row.Epoch, &row.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return ActivationRow{Owner: owner, PluginID: pluginID, Enabled: false, Epoch: 0}, nil + } + if err != nil { + return row, err + } + row.Enabled = enabled != 0 + return row, nil +} + +// SetActivation 事务内读-改-写：每次状态变更（含同值重设）递增 epoch， +// 使禁用/启用前的在途提交按 epoch 失配被拒（KTD7 禁用是准入截止点）。 +func (s *Store) SetActivation(owner, pluginID string, enabled bool) (ActivationRow, error) { + tx, err := s.db.Begin() + if err != nil { + return ActivationRow{}, err + } + defer tx.Rollback() + now := time.Now().UTC().Format(time.RFC3339) + var epoch int64 + err = tx.QueryRow(s.q(`SELECT epoch FROM plugin_activation WHERE owner=? AND plugin_id=?`), owner, pluginID).Scan(&epoch) + if errors.Is(err, sql.ErrNoRows) { + epoch = 0 + } else if err != nil { + return ActivationRow{}, err + } + epoch++ + flag := 0 + if enabled { + flag = 1 + } + if _, err := tx.Exec(s.q(`INSERT INTO plugin_activation(owner,plugin_id,enabled,epoch,updated_at) VALUES(?,?,?,?,?) + ON CONFLICT(owner, plugin_id) DO UPDATE SET enabled=excluded.enabled, epoch=excluded.epoch, updated_at=excluded.updated_at`), + owner, pluginID, flag, epoch, now); err != nil { + return ActivationRow{}, err + } + if err := tx.Commit(); err != nil { + return ActivationRow{}, err + } + return ActivationRow{Owner: owner, PluginID: pluginID, Enabled: enabled, Epoch: epoch, UpdatedAt: now}, nil +} + +// SaveSurfaceSnapshot 保存一个新修订（乐观并发）：baseRevision=0 表示新建。 +// 身份信封字段以参数为准（服务端绑定），不采信快照内自报值；返回带最终 +// revision 的快照（CAS 命中后的实际版本号）。 +func (s *Store) SaveSurfaceSnapshot(snap core.SurfaceSnapshot) (core.SurfaceSnapshot, error) { + tx, err := s.db.Begin() + if err != nil { + return snap, err + } + defer tx.Rollback() + now := time.Now().UTC().Format(time.RFC3339) + specJSON, err := json.Marshal(snap.Spec) + if err != nil { + return snap, err + } + if len(specJSON) > core.SurfaceMaxBytes { + return snap, fmt.Errorf("surface: spec %d bytes exceeds %d limit", len(specJSON), core.SurfaceMaxBytes) + } + compatJSON, _ := json.Marshal(snap.Compat) + var latest int + err = tx.QueryRow(s.q(`SELECT latest_revision FROM surfaces WHERE surface_id=? AND owner=?`), + string(snap.Identity.SurfaceID), snap.Identity.Owner).Scan(&latest) + switch { + case errors.Is(err, sql.ErrNoRows): + if snap.BaseRevision != 0 { + return snap, &ErrRevisionConflict{Current: 0} + } + if _, err := tx.Exec(s.q(`INSERT INTO surfaces(surface_id,owner,project_id,conversation_id,plugin_id,compat,latest_revision,created_at,updated_at) + VALUES(?,?,?,?,?,?,1,?,?)`), + string(snap.Identity.SurfaceID), snap.Identity.Owner, string(snap.Identity.ProjectID), + string(snap.Identity.ConversationID), snap.Identity.PluginID, string(compatJSON), now, now); err != nil { + return snap, err + } + snap.Revision = 1 + case err != nil: + return snap, err + default: + if snap.BaseRevision != latest { + return snap, &ErrRevisionConflict{Current: latest} + } + snap.Revision = latest + 1 + if _, err := tx.Exec(s.q(`UPDATE surfaces SET latest_revision=?, updated_at=? WHERE surface_id=? AND owner=?`), + snap.Revision, now, string(snap.Identity.SurfaceID), snap.Identity.Owner); err != nil { + return snap, err + } + } + if _, err := tx.Exec(s.q(`INSERT INTO surface_revisions(surface_id,revision,base_revision,context_version,activation_epoch,spec,semantic_text,event_pending,created_at) + VALUES(?,?,?,?,?,?,?,?,?)`), + string(snap.Identity.SurfaceID), snap.Revision, snap.BaseRevision, snap.ContextVersion, + snap.ActivationEpoch, string(specJSON), snap.SemanticText, 0, now); err != nil { + return snap, err + } + return snap, tx.Commit() +} + +func (s *Store) GetSurface(owner, surfaceID string) (SurfaceRecord, error) { + var r SurfaceRecord + var compat string + err := s.db.QueryRow(s.q(`SELECT surface_id,owner,project_id,conversation_id,plugin_id,compat,latest_revision,created_at,updated_at + FROM surfaces WHERE surface_id=? AND owner=?`), surfaceID, owner). + Scan(&r.SurfaceID, &r.Owner, &r.ProjectID, &r.ConversationID, &r.PluginID, &compat, &r.LatestRevision, &r.CreatedAt, &r.UpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return r, ErrNotFound + } + r.Compat = compat + return r, err +} + +// GetSurfaceSnapshot 取整页快照；revision=0 表示最新。 +func (s *Store) GetSurfaceSnapshot(owner, surfaceID string, revision int) (core.SurfaceSnapshot, error) { + rec, err := s.GetSurface(owner, surfaceID) + if err != nil { + return core.SurfaceSnapshot{}, err + } + if revision <= 0 { + revision = rec.LatestRevision + } + var row SurfaceRevisionRow + var pending int + err = s.db.QueryRow(s.q(`SELECT surface_id,revision,base_revision,context_version,activation_epoch,spec,semantic_text,event_pending,created_at + FROM surface_revisions WHERE surface_id=? AND revision=?`), surfaceID, revision). + Scan(&row.SurfaceID, &row.Revision, &row.BaseRevision, &row.ContextVersion, &row.ActivationEpoch, &row.Spec, &row.SemanticText, &pending, &row.CreatedAt) + if errors.Is(err, sql.ErrNoRows) { + return core.SurfaceSnapshot{}, ErrNotFound + } + if err != nil { + return core.SurfaceSnapshot{}, err + } + row.EventPending = pending != 0 + var spec core.SurfaceSpec + if err := json.Unmarshal([]byte(row.Spec), &spec); err != nil { + return core.SurfaceSnapshot{}, fmt.Errorf("surface: corrupted spec row: %w", err) + } + var compat core.SurfaceCompat + _ = json.Unmarshal([]byte(rec.Compat), &compat) + return core.SurfaceSnapshot{ + Identity: core.SurfaceIdentity{ + SurfaceID: core.ID(rec.SurfaceID), + Owner: rec.Owner, + ProjectID: core.ID(rec.ProjectID), + ConversationID: core.ID(rec.ConversationID), + PluginID: rec.PluginID, + }, + Compat: compat, + Revision: row.Revision, + BaseRevision: row.BaseRevision, + ContextVersion: row.ContextVersion, + ActivationEpoch: row.ActivationEpoch, + Spec: spec, + SemanticText: row.SemanticText, + }, nil +} + +func (s *Store) ListSurfaces(owner, conversationID string) ([]core.SurfaceMeta, error) { + // 最新修订行的 spec 在 Go 侧解析出 title（json_extract 为 SQLite 方言，PG 不可用）。 + rows, err := s.db.Query(s.q(`SELECT s.surface_id, s2.spec, s.plugin_id, s.latest_revision, s.conversation_id, s.updated_at + FROM surfaces s + JOIN surface_revisions s2 ON s2.surface_id=s.surface_id AND s2.revision=s.latest_revision + WHERE s.owner=? AND (?='' OR s.conversation_id=?) ORDER BY s.updated_at DESC`), + owner, conversationID, conversationID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []core.SurfaceMeta + for rows.Next() { + var m core.SurfaceMeta + var specJSON string + if err := rows.Scan(&m.SurfaceID, &specJSON, &m.PluginID, &m.Revision, &m.ConversationID, &m.UpdatedAt); err != nil { + return nil, err + } + var spec core.SurfaceSpec + _ = json.Unmarshal([]byte(specJSON), &spec) + m.Title = spec.Title + out = append(out, m) + } + return out, rows.Err() +} + +// SurfaceOwner 反查页面归属（待恢复扫描行不带 owner）。 +func (s *Store) SurfaceOwner(surfaceID string) (string, error) { + var owner string + err := s.db.QueryRow(s.q(`SELECT owner FROM surfaces WHERE surface_id=?`), surfaceID).Scan(&owner) + if errors.Is(err, sql.ErrNoRows) { + return "", ErrNotFound + } + return owner, err +} + +// SetSurfaceEventPending 标记/清除修订行的投影事件待恢复态（写快照后事件发布失败， +// KTD「持久事件或重建快照恢复」；恢复扫描走 ListSurfaceEventsPending）。 +func (s *Store) SetSurfaceEventPending(owner, surfaceID string, revision int, pending bool) error { + flag := 0 + if pending { + flag = 1 + } + res, err := s.db.Exec(s.q(`UPDATE surface_revisions SET event_pending=? + WHERE surface_id=? AND revision=? AND EXISTS(SELECT 1 FROM surfaces WHERE surface_id=? AND owner=?)`), + flag, surfaceID, revision, surfaceID, owner) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return ErrNotFound + } + return nil +} + +// ListSurfaceEventsPending 扫描待恢复的修订行（全租户，恢复进程专用）。 +func (s *Store) ListSurfaceEventsPending(limit int) ([]SurfaceRevisionRow, error) { + if limit <= 0 { + limit = 100 + } + rows, err := s.db.Query(s.q(`SELECT surface_id,revision,base_revision,context_version,activation_epoch,spec,semantic_text,event_pending,created_at + FROM surface_revisions WHERE event_pending=1 ORDER BY created_at LIMIT ?`), limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []SurfaceRevisionRow + for rows.Next() { + var r SurfaceRevisionRow + var pending int + if err := rows.Scan(&r.SurfaceID, &r.Revision, &r.BaseRevision, &r.ContextVersion, &r.ActivationEpoch, &r.Spec, &r.SemanticText, &pending, &r.CreatedAt); err != nil { + return nil, err + } + r.EventPending = pending != 0 + out = append(out, r) + } + return out, rows.Err() +} diff --git a/internal/store/surfaces_test.go b/internal/store/surfaces_test.go new file mode 100644 index 0000000..bfd586e --- /dev/null +++ b/internal/store/surfaces_test.go @@ -0,0 +1,126 @@ +package store + +import ( + "testing" + + core "agent-system-core" +) + +func testSnapshot(surfaceID, owner, plugin string, base int) core.SurfaceSnapshot { + return core.SurfaceSnapshot{ + Identity: core.SurfaceIdentity{SurfaceID: core.ID(surfaceID), Owner: owner, PluginID: plugin, ConversationID: "c1"}, + BaseRevision: base, + ActivationEpoch: 1, + Spec: core.SurfaceSpec{Title: "t", Nodes: []core.UINode{{Type: core.UIMarkdown, ID: "n1", Markdown: "x"}}}, + } +} + +func TestSurfaceSaveCASAndRead(t *testing.T) { + st := openTestStore(t) + // 新建 → revision 1 + snap, err := st.SaveSurfaceSnapshot(testSnapshot("s1", "alice", "genui", 0)) + if err != nil || snap.Revision != 1 { + t.Fatalf("create: rev=%d err=%v", snap.Revision, err) + } + // 基于最新版的更新 → revision 2 + snap2, err := st.SaveSurfaceSnapshot(testSnapshot("s1", "alice", "genui", 1)) + if err != nil || snap2.Revision != 2 { + t.Fatalf("update: rev=%d err=%v", snap2.Revision, err) + } + // 过期 base → 冲突并携带当前版本 + _, err = st.SaveSurfaceSnapshot(testSnapshot("s1", "alice", "genui", 1)) + var conflict *ErrRevisionConflict + if err == nil { + t.Fatal("stale base should conflict") + } + if e, ok := err.(*ErrRevisionConflict); !ok { + t.Fatalf("want ErrRevisionConflict, got %v", err) + } else if e.Current != 2 { + t.Fatalf("conflict should carry current=2, got %d", e.Current) + } + _ = conflict + // owner 隔离：bob 读不到 alice 的页面 + if _, err := st.GetSurfaceSnapshot("bob", "s1", 0); err != ErrNotFound { + t.Fatalf("cross-owner read should be NotFound, got %v", err) + } + // 跨 owner 写同 id → 冲突（surface 已存在且非本 owner 视为冲突路径） + if _, err := st.SaveSurfaceSnapshot(testSnapshot("s1", "bob", "genui", 0)); err == nil { + t.Fatal("cross-owner create on existing id should fail") + } + // 读取往返：字段保真 + got, err := st.GetSurfaceSnapshot("alice", "s1", 0) + if err != nil || got.Revision != 2 || got.Spec.Title != "t" || len(got.Spec.Nodes) != 1 { + t.Fatalf("read-back mismatch: %+v err=%v", got, err) + } + old, err := st.GetSurfaceSnapshot("alice", "s1", 1) + if err != nil || old.Revision != 1 { + t.Fatalf("historical revision read failed: %+v err=%v", old, err) + } +} + +func TestSurfaceActivationEpoch(t *testing.T) { + st := openTestStore(t) + // 无行 = 默认关闭（KTD7：不沿用 grants 缺省 true） + a, err := st.GetActivation("alice", "genui") + if err != nil || a.Enabled || a.Epoch != 0 { + t.Fatalf("default should be disabled/epoch0: %+v err=%v", a, err) + } + a1, err := st.SetActivation("alice", "genui", true) + if err != nil || !a1.Enabled || a1.Epoch != 1 { + t.Fatalf("enable: %+v err=%v", a1, err) + } + a2, err := st.SetActivation("alice", "genui", false) + if err != nil || a2.Enabled || a2.Epoch != 2 { + t.Fatalf("disable should bump epoch: %+v err=%v", a2, err) + } + a3, err := st.SetActivation("alice", "genui", true) + if err != nil || a3.Epoch != 3 { + t.Fatalf("re-enable should bump epoch: %+v err=%v", a3, err) + } + // owner 隔离 + b, _ := st.GetActivation("bob", "genui") + if b.Enabled || b.Epoch != 0 { + t.Fatalf("bob activation leaked: %+v", b) + } +} + +func TestSurfacePendingEvents(t *testing.T) { + st := openTestStore(t) + if _, err := st.SaveSurfaceSnapshot(testSnapshot("s2", "alice", "genui", 0)); err != nil { + t.Fatal(err) + } + if err := st.SetSurfaceEventPending("alice", "s2", 1, true); err != nil { + t.Fatal(err) + } + pend, err := st.ListSurfaceEventsPending(10) + if err != nil || len(pend) != 1 || pend[0].SurfaceID != "s2" || pend[0].Revision != 1 { + t.Fatalf("pending scan: %+v err=%v", pend, err) + } + // owner 校验：他人不能标记/清除 + if err := st.SetSurfaceEventPending("bob", "s2", 1, false); err != ErrNotFound { + t.Fatalf("cross-owner pending clear should be NotFound, got %v", err) + } + if err := st.SetSurfaceEventPending("alice", "s2", 1, false); err != nil { + t.Fatal(err) + } + if pend, _ := st.ListSurfaceEventsPending(10); len(pend) != 0 { + t.Fatalf("pending should be cleared: %+v", pend) + } + // 列表投影（含 title 解析） + if _, err := st.SaveSurfaceSnapshot(testSnapshot("s3", "alice", "genui", 0)); err != nil { + t.Fatal(err) + } + metas, err := st.ListSurfaces("alice", "c1") + if err != nil || len(metas) != 2 { + t.Fatalf("list: %+v err=%v", metas, err) + } + for _, m := range metas { + if m.Title != "t" { + t.Fatalf("meta title not parsed: %+v", m) + } + } + other, _ := st.ListSurfaces("alice", "c-other") + if len(other) != 0 { + t.Fatalf("conversation filter broken: %+v", other) + } +} diff --git a/internal/surfacesvc/surfacesvc.go b/internal/surfacesvc/surfacesvc.go new file mode 100644 index 0000000..b6fe1af --- /dev/null +++ b/internal/surfacesvc/surfacesvc.go @@ -0,0 +1,202 @@ +// Package surfacesvc 是 Generative UI Surface 的服务层（提案 U2）： +// 提交校验（目录纪律）→ 启禁/epoch 准入 → owner 隔离存储（revision CAS）→ +// 事件发布（失败转待恢复）。所有入口（插件命令句柄、直接 API）共用本层， +// 守卫不依赖入口自觉（KTD7：执行端统一守卫，禁用是准入截止点）。 +package surfacesvc + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "log" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// Emitter 由宿主注入（与 cardsvc 同一事件管道）：持久化、seq 排序、断线重放。 +type Emitter func(conversationID, runID string, typ core.StreamEventType, payload any) error + +var ( + ErrNotFound = store.ErrNotFound + ErrDisabled = errors.New("surface: plugin not enabled for owner") + ErrStaleEpoch = errors.New("surface: activation epoch mismatch (stale in-flight submit)") + ErrUnavailable = errors.New("surface: capability not available") +) + +// ErrInvalidSpec 携带字段级违规清单（目录=提示=校验三同源；供原 Agent 修复重提）。 +type ErrInvalidSpec struct{ Errs []string } + +func (e *ErrInvalidSpec) Error() string { return "surface: invalid spec: " + fmt.Sprint(e.Errs) } + +// SubmitRequest 是一次页面提交（工具参数通道或直接 API 同形）。 +// ObservedEpoch>0 时校验与当前 epoch 一致，拦截「禁用后迟到的在途结果」。 +type SubmitRequest struct { + SurfaceID core.ID + ConversationID core.ID + SourceRunID core.ID + BaseRevision int + ObservedEpoch int64 + ContextVersion string + SemanticText string + Spec json.RawMessage +} + +type Service struct { + st *store.Store + emit Emitter + now func() (t string) + logln func(format string, v ...any) +} + +func New(st *store.Store, emit Emitter) *Service { + return &Service{st: st, emit: emit, + now: func() string { return "" }, + logln: log.Printf} +} + +// Submit 是唯一的写入咽喉：严格解码 → 目录校验 → 启禁/epoch → CAS 保存 → 事件。 +// 写入成功但事件发布失败时标记待恢复并照常返回快照（可恢复的最终一致）。 +func (s *Service) Submit(ctx context.Context, owner, pluginID string, req SubmitRequest) (core.SurfaceSnapshot, error) { + if owner == "" || pluginID == "" { + return core.SurfaceSnapshot{}, ErrUnavailable + } + act, err := s.st.GetActivation(owner, pluginID) + if err != nil { + return core.SurfaceSnapshot{}, err + } + if !act.Enabled { + return core.SurfaceSnapshot{}, ErrDisabled + } + if req.ObservedEpoch > 0 && req.ObservedEpoch != act.Epoch { + return core.SurfaceSnapshot{}, ErrStaleEpoch + } + spec, err := decodeSpecStrict(req.Spec) + if err != nil { + return core.SurfaceSnapshot{}, &ErrInvalidSpec{Errs: []string{"spec: " + err.Error()}} + } + if errs := core.ValidateSurfaceSpec(spec, nil); len(errs) > 0 { + // knownSources=nil：来源授权属 U3 数据绑定层，本层只做格式与纪律检查。 + return core.SurfaceSnapshot{}, &ErrInvalidSpec{Errs: errs} + } + surfaceID := req.SurfaceID + if surfaceID == "" { + surfaceID = core.ID(store.NewID("surf")) + } + snap := core.SurfaceSnapshot{ + Identity: core.SurfaceIdentity{ + SurfaceID: surfaceID, Owner: owner, ConversationID: req.ConversationID, + SourceRunID: req.SourceRunID, PluginID: pluginID, + }, + Compat: core.SurfaceCompat{SchemaVersion: core.SurfaceSchemaVersion, CatalogVersion: core.SurfaceCatalogVersion}, + BaseRevision: req.BaseRevision, + ContextVersion: req.ContextVersion, + ActivationEpoch: act.Epoch, + Spec: spec, + SemanticText: req.SemanticText, + } + saved, err := s.st.SaveSurfaceSnapshot(snap) + if err != nil { + return saved, err + } + s.publish(saved, false) + return saved, nil +} + +// Get 读取整页快照（owner 隔离）。禁用不阻断读取：历史页面按权限保留（KTD7）。 +func (s *Service) Get(ctx context.Context, owner string, surfaceID core.ID) (core.SurfaceSnapshot, error) { + return s.st.GetSurfaceSnapshot(owner, string(surfaceID), 0) +} + +// List 列出 owner 名下（可按会话过滤）的页面元数据。 +func (s *Service) List(ctx context.Context, owner string, conversationID core.ID) ([]core.SurfaceMeta, error) { + return s.st.ListSurfaces(owner, string(conversationID)) +} + +// Activation 观察当前启禁与 epoch（调用方装配能力时快照 ObservedEpoch 用）。 +func (s *Service) Activation(ctx context.Context, owner, pluginID string) (store.ActivationRow, error) { + return s.st.GetActivation(owner, pluginID) +} + +// SetActivation 启/禁 Surface 能力：先持久化并递增 epoch，再返回（禁用返回即 +// 准入截止点）；审计留痕。新插件默认关闭（无行=disabled）。 +func (s *Service) SetActivation(ctx context.Context, owner, pluginID string, enabled bool) (store.ActivationRow, error) { + row, err := s.st.SetActivation(owner, pluginID, enabled) + if err != nil { + return row, err + } + action := "surface.disabled" + if enabled { + action = "surface.enabled" + } + if err := s.st.AppendAuditEvent(store.AuditEvent{ + Owner: owner, Actor: "host", Action: action, + Target: pluginID, Evidence: fmt.Sprintf("epoch=%d", row.Epoch), + }); err != nil { + return row, err + } + return row, nil +} + +// publish 发布投影事件；失败则标记待恢复（快照已落盘，不回滚写入）。 +func (s *Service) publish(snap core.SurfaceSnapshot, recovery bool) { + if s.emit == nil || snap.Identity.ConversationID == "" { + return + } + if err := s.emit(string(snap.Identity.ConversationID), string(snap.Identity.SourceRunID), core.EvSurfaceUpdated, snap); err != nil { + if markErr := s.st.SetSurfaceEventPending(snap.Identity.Owner, string(snap.Identity.SurfaceID), snap.Revision, true); markErr != nil { + s.logln("surface: event publish failed (%v) and pending-mark failed (%v) rev=%d", err, markErr, snap.Revision) + return + } + if recovery { + s.logln("surface: recovery emit failed again, kept pending rev=%d", snap.Revision) + } + } +} + +// RecoverPendingEvents 重发待恢复的投影事件；返回恢复条数。 +func (s *Service) RecoverPendingEvents(ctx context.Context) (int, error) { + rows, err := s.st.ListSurfaceEventsPending(100) + if err != nil { + return 0, err + } + n := 0 + for _, row := range rows { + owner, err := s.st.SurfaceOwner(row.SurfaceID) + if err != nil { + continue + } + snap, err := s.st.GetSurfaceSnapshot(owner, row.SurfaceID, row.Revision) + if err != nil { + continue + } + s.publish(snap, true) + if err := s.st.SetSurfaceEventPending(owner, row.SurfaceID, row.Revision, false); err != nil { + continue + } + n++ + } + return n, nil +} + +// decodeSpecStrict 严格解码：未知字段拒绝（DisallowUnknownFields），并限制体积。 +func decodeSpecStrict(raw json.RawMessage) (core.SurfaceSpec, error) { + if len(raw) == 0 { + return core.SurfaceSpec{}, errors.New("empty spec") + } + if len(raw) > core.SurfaceMaxBytes { + return core.SurfaceSpec{}, fmt.Errorf("spec %d bytes exceeds %d limit", len(raw), core.SurfaceMaxBytes) + } + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + var spec core.SurfaceSpec + if err := dec.Decode(&spec); err != nil { + return core.SurfaceSpec{}, err + } + if dec.More() { + return core.SurfaceSpec{}, errors.New("trailing data after spec") + } + return spec, nil +} diff --git a/internal/surfacesvc/surfacesvc_test.go b/internal/surfacesvc/surfacesvc_test.go new file mode 100644 index 0000000..9f615d0 --- /dev/null +++ b/internal/surfacesvc/surfacesvc_test.go @@ -0,0 +1,148 @@ +package surfacesvc + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "testing" + + "agent-system-backend/internal/store" + core "agent-system-core" +) + +func openSvcStore(t *testing.T) *store.Store { + t.Helper() + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +func validSpecJSON() json.RawMessage { + return json.RawMessage(`{"title":"选型对比","nodes":[ + {"type":"section","id":"n1","title":"概览","children":[ + {"type":"markdown","id":"n1a","source":["doc:a"],"markdown":"结论先行。"}]}, + {"type":"table","id":"n2","source":["doc:a","doc:b"],"table":{ + "columns":[{"key":"k","label":"维度"}], + "rows":[{"cells":[["k","价格"]]}]}}]}`) +} + +func newSvc(t *testing.T) (*Service, *[]core.StreamEventType) { + st := openSvcStore(t) + var events []core.StreamEventType + return New(st, func(conversationID, runID string, typ core.StreamEventType, payload any) error { + events = append(events, typ) + return nil + }), &events +} + +func TestSubmitGuards(t *testing.T) { + ctx := context.Background() + svc, _ := newSvc(t) + + // 默认关闭：新插件无 activation 行 → 拒绝 + if _, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{Spec: validSpecJSON()}); !errors.Is(err, ErrDisabled) { + t.Fatalf("default-disabled should reject, got %v", err) + } + // 启用后提交成功 + if _, err := svc.SetActivation(ctx, "alice", "genui", true); err != nil { + t.Fatal(err) + } + snap, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{ConversationID: "c1", Spec: validSpecJSON()}) + if err != nil || snap.Revision != 1 || snap.Identity.Owner != "alice" || snap.Identity.PluginID != "genui" { + t.Fatalf("submit: %+v err=%v", snap, err) + } + // epoch 失配：观察值与当前不一致（禁用/重启用后的在途结果）→ 拒绝 + if _, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{ + SurfaceID: snap.Identity.SurfaceID, BaseRevision: 1, ObservedEpoch: 99, Spec: validSpecJSON(), + }); !errors.Is(err, ErrStaleEpoch) { + t.Fatalf("stale epoch should reject, got %v", err) + } + // 规格违规：未知类型 + 字段级清单 + bad := json.RawMessage(`{"title":"x","nodes":[{"type":"hologram"}]}`) + _, err = svc.Submit(ctx, "alice", "genui", SubmitRequest{Spec: bad}) + var inv *ErrInvalidSpec + if !errors.As(err, &inv) || len(inv.Errs) == 0 { + t.Fatalf("invalid spec should carry field errors, got %v", err) + } + // 严格解码：未知字段拒绝（目录纪律） + extra := json.RawMessage(`{"title":"x","nodes":[{"type":"list","items":["a"],"color":"red"}]}`) + if _, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{Spec: extra}); err == nil { + t.Fatal("unknown field should be rejected") + } + // 禁用后：提交拒绝、读取保留（KTD7 历史页面不丢） + if _, err := svc.SetActivation(ctx, "alice", "genui", false); err != nil { + t.Fatal(err) + } + if _, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{Spec: validSpecJSON()}); !errors.Is(err, ErrDisabled) { + t.Fatalf("disabled submit should reject, got %v", err) + } + if _, err := svc.Get(ctx, "alice", snap.Identity.SurfaceID); err != nil { + t.Fatalf("disabled should not block reads, got %v", err) + } + // owner 隔离 + if _, err := svc.Get(ctx, "bob", snap.Identity.SurfaceID); !errors.Is(err, ErrNotFound) { + t.Fatalf("cross-owner get should be NotFound, got %v", err) + } +} + +func TestSubmitConflictSurfacesCurrent(t *testing.T) { + ctx := context.Background() + svc, _ := newSvc(t) + svc.SetActivation(ctx, "alice", "genui", true) + s1, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{ConversationID: "c1", Spec: validSpecJSON()}) + if err != nil { + t.Fatal(err) + } + s2, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{ + SurfaceID: s1.Identity.SurfaceID, BaseRevision: 1, ConversationID: "c1", Spec: validSpecJSON(), + }) + if err != nil || s2.Revision != 2 { + t.Fatalf("concurrent second writer: %+v err=%v", s2, err) + } + // 用过期 base 的第三方写入 → 冲突 + _, err = svc.Submit(ctx, "alice", "genui", SubmitRequest{ + SurfaceID: s1.Identity.SurfaceID, BaseRevision: 1, ConversationID: "c1", Spec: validSpecJSON(), + }) + var conflict *store.ErrRevisionConflict + if !errors.As(err, &conflict) || conflict.Current != 2 { + t.Fatalf("conflict should surface current=2, got %v", err) + } +} + +func TestEventPublishFailureRecovery(t *testing.T) { + ctx := context.Background() + st := openSvcStore(t) + fail := true + var emitted int + svc := New(st, func(conversationID, runID string, typ core.StreamEventType, payload any) error { + if fail { + return errors.New("bus down") + } + emitted++ + return nil + }) + svc.SetActivation(ctx, "alice", "genui", true) + // 快照保存成功、事件发布失败 → 待恢复标记 + snap, err := svc.Submit(ctx, "alice", "genui", SubmitRequest{ConversationID: "c1", Spec: validSpecJSON()}) + if err != nil { + t.Fatalf("snapshot must survive event failure, got %v", err) + } + pend, _ := st.ListSurfaceEventsPending(10) + if len(pend) != 1 { + t.Fatalf("event should be pending, got %d", len(pend)) + } + // 恢复：事件通道恢复后重发并清除 + fail = false + n, err := svc.RecoverPendingEvents(ctx) + if err != nil || n != 1 || emitted != 1 { + t.Fatalf("recovery: n=%d emitted=%d err=%v", n, emitted, err) + } + if pend, _ := st.ListSurfaceEventsPending(10); len(pend) != 0 { + t.Fatalf("pending should clear, got %d", len(pend)) + } + _ = snap +} diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index cd208eb..37bad19 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -3,6 +3,7 @@ package extension import ( + "context" "encoding/json" "errors" "sync" @@ -21,7 +22,10 @@ import ( // 2.5.0-experimental：Options.DisableExecTools（执行原语启动级硬关闭）。 // 2.6.0-experimental：Options.ExecBackend（EXT-16 执行后端 direct|sandbox）。 // 2.7.0-experimental：Registry.AddIntent 与 IntentSpec（G2-2 意图声明面）。 -const SurfaceVersion = "2.7.0-experimental" +// 2.8.0-experimental：EXT-17 新增可选能力接口 SurfaceHost（SurfaceSave/Get/List/ +// Activation；Generative UI U2）。实现由基座 hostcaps 注入；未装配时类型断言 +// 不成立即明确拒绝，不给必实现 Host 接口加方法（不破坏既有消费者）。 +const SurfaceVersion = "2.8.0-experimental" // ErrNotLinked 表示进程没有空白导入 link 包，Main 无法启动服务。 var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent-system-backend/pkg/extension/link") @@ -47,6 +51,35 @@ type Registry interface { AddIntent(spec json.RawMessage) error } +// SurfaceSaveRequest 是一次页面提交（EXT-17）：Spec 为 core.SurfaceSpec 的 +// 严格 JSON（未知字段拒绝）；身份信封由宿主按分发 Owner/PluginID 绑定， +// 不采信请求内自报的 owner/pluginId。 +type SurfaceSaveRequest struct { + SurfaceID core.ID `json:"surfaceId,omitempty"` + ConversationID core.ID `json:"conversationId,omitempty"` + SourceRunID core.ID `json:"sourceRunId,omitempty"` + BaseRevision int `json:"baseRevision,omitempty"` + ObservedEpoch int64 `json:"observedEpoch,omitempty"` + ContextVersion string `json:"contextVersion,omitempty"` + SemanticText string `json:"semanticText,omitempty"` + Spec json.RawMessage `json:"spec"` +} + +// SurfaceHost 是 EXT-17 可选能力接口：插件经宿主保存/读取生成页面。 +// 通过对分发 Host 句柄做类型断言获取（断言失败=能力缺失，明确报错不降级）。 +type SurfaceHost interface { + // SurfaceSave 校验并保存页面；禁用/epoch 失配/规格违规/revision 冲突 + // 分别返回确定错误，携带字段级清单时调用方可修复后至多重提一次。 + SurfaceSave(ctx context.Context, req SurfaceSaveRequest) (core.SurfaceSnapshot, error) + // SurfaceGet 读取整页快照（owner 作用域；历史页面禁用后仍可读）。 + SurfaceGet(ctx context.Context, surfaceID core.ID) (core.SurfaceSnapshot, error) + // SurfaceList 列出当前 owner 名下（可按会话过滤）的页面元数据。 + SurfaceList(ctx context.Context, conversationID core.ID) ([]core.SurfaceMeta, error) + // SurfaceActivation 观察本插件在当前 owner 的启禁与 epoch（装配能力时 + // 快照 ObservedEpoch，拦截禁用后迟到的在途提交）。 + SurfaceActivation(ctx context.Context) (enabled bool, epoch int64, err error) +} + // IntentSpec 意图声明规格（G2-2）：id 形如 .；phrases 是触发 // 短语（宿主归一后包含匹配，最长短语优先）；command 须为已注册命令 id； // params 为参数模板（调用方可覆盖）。宿主侧路由入口：system.intent 命令。 diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index 36961b1..666e4a0 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -37,4 +37,10 @@ ResetCompiled ResourceAuthorizer SetRunner SetScopedHostFactory +SurfaceHost +SurfaceHost.SurfaceActivation +SurfaceHost.SurfaceGet +SurfaceHost.SurfaceList +SurfaceHost.SurfaceSave +SurfaceSaveRequest SurfaceVersion