From cb26e4c9473f75ad015a83e3bba8137a08ff56f9 Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 5 Oct 2026 11:39:25 +0000 Subject: [PATCH] Add EXT-1 inproc plugin command handler with owner-scoped host handle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: 进程内插件命令此前只登记规格不绑处理器，分发即 "no host handler bound"， 且插件无 owner 凭证（H4）；EXT-1 最小集补上处理器绑定与 owner 作用域宿主句柄。 Evidence: pkg/extension 新增 Request/Handler/Host + Registry.AddCommandHandler（surface.txt 12→15）； cmdbus.RegisterPluginCommand 三例测试（committed/未绑 rejected/高危确认卡先拦）； rag.IndexChunks 抽出并两例直测；internal/app hostcaps+inprocHost 复用 hostapi owner 作用域实现， applyCompiled 携 caps，inproc_host_test 双 owner 隔离 + provider=hash 向量写入均通过；go test ./... 29 包全绿。 Impact: 公开面依赖从零外部模块变为 +agent-system-core（声明式 go.mod 需按 compat-policy §4 加 core require+replace）； 既有路径语义不变，SurfaceVersion 保持 2.0.0-experimental。 Refs: documents 提案 2026-10-05-ext-1-inproc-command-handler.md；V1.2 方案批注 1。 --- internal/app/app.go | 12 +- internal/app/compiled.go | 36 +++++- internal/app/hostcaps.go | 116 ++++++++++++++++++ internal/app/inproc_host_test.go | 195 +++++++++++++++++++++++++++++++ internal/cmdbus/cmdbus.go | 12 ++ internal/cmdbus/cmdbus_test.go | 69 +++++++++++ internal/rag/hybrid.go | 30 ++++- internal/rag/rag_test.go | 65 +++++++++++ pkg/extension/extension.go | 3 + pkg/extension/host.go | 55 +++++++++ pkg/extension/surface.txt | 3 + 11 files changed, 581 insertions(+), 15 deletions(-) create mode 100644 internal/app/hostcaps.go create mode 100644 internal/app/inproc_host_test.go create mode 100644 pkg/extension/host.go diff --git a/internal/app/app.go b/internal/app/app.go index 1c4931f..246aa47 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -344,7 +344,9 @@ func Run(opt Options) error { bus := cmdbus.New(st, cardSvc, func() time.Time { return time.Now().UTC() }) bus.SetPendingStore(st) pluginHost.OnCommands = bus.RegisterPluginCommands - applyCompiled(skillsRoot, bus) + // hostcaps：hostapi 闭包与进程内插件句柄共用的 owner 作用域能力（EXT-1 批注 1）。 + caps := hostcaps{st: st, pipe: pipe, set: setSvc} + applyCompiled(skillsRoot, bus, caps) mgr.OnUsage = func(owner string, tokens int) { budget := int(setSvc.Effective(owner, "").GetNumber("llm.dailyTokenBudget", 0)) status, aerr := st.AccountTokens(owner, tokens, budget) @@ -365,14 +367,10 @@ func Run(opt Options) error { return nil } hostSrv.Search = func(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { - return pipe.Retrieve(ctx, core.ID(owner), core.RetrievalQuery{Text: text}) + return caps.searchDefault(ctx, owner, text) } hostSrv.Complete = func(ctx context.Context, owner, prompt string) (string, error) { - route := runtime.ResolveChatModel(core.AgentProfile{}, setSvc.Effective(owner, "")) - if route.Model == "" || route.Base == "" { - return "", fmt.Errorf("llm route not configured") - } - return completeChat(ctx, route.Base, route.Key, route.Model, prompt) + return caps.complete(ctx, owner, prompt) } srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost) diff --git a/internal/app/compiled.go b/internal/app/compiled.go index f35d656..c38a76e 100644 --- a/internal/app/compiled.go +++ b/internal/app/compiled.go @@ -1,11 +1,14 @@ package app import ( + "context" "encoding/json" + "fmt" "log" "os" "path/filepath" + agentcommand "agent-command" "agent-system-backend/internal/cmdbus" "agent-system-backend/pkg/extension" core "agent-system-core" @@ -16,6 +19,34 @@ type compiledRegistry struct { bus *cmdbus.Bus } +// pluginRegistry 携带插件身份与宿主能力，per-InProc 构造： +// AddCommandHandler 绑定的处理器在分发时按请求 Owner 构造 inprocHost（EXT-1）。 +type pluginRegistry struct { + compiledRegistry + pluginID string + caps hostcaps +} + +func (r pluginRegistry) AddCommandHandler(spec json.RawMessage, h extension.Handler) error { + if r.bus == nil { + return nil + } + if h == nil { + return fmt.Errorf("extension: %s: nil command handler", r.pluginID) + } + var s agentcommand.CommandSpec + if err := json.Unmarshal(spec, &s); err != nil { + return fmt.Errorf("extension: command spec: %w", err) + } + return r.bus.RegisterPluginCommand(s, func(ctx context.Context, req cmdbus.Request) (any, error) { + host := inprocHost{caps: r.caps, pluginID: r.pluginID, owner: req.Owner} + return h(ctx, host, extension.Request{ + ID: req.ID, Params: req.Params, Entrance: req.Entrance, + Actor: req.Actor, Owner: req.Owner, ConversationID: req.ConversationID, + }) + }) +} + func (r compiledRegistry) AddSkill(name, markdown string) error { if r.skillsRoot == "" || name == "" { return nil @@ -39,12 +70,13 @@ func (r compiledRegistry) Note(kind, name string, payload json.RawMessage) error return nil } -func applyCompiled(skillsRoot string, bus *cmdbus.Bus) { - reg := compiledRegistry{skillsRoot: skillsRoot, bus: bus} +func applyCompiled(skillsRoot string, bus *cmdbus.Bus, caps hostcaps) { + base := compiledRegistry{skillsRoot: skillsRoot, bus: bus} for _, e := range extension.Compiled() { if e.Register == nil { continue } + reg := pluginRegistry{compiledRegistry: base, pluginID: e.ID, caps: caps} if err := e.Register(reg); err != nil { log.Printf("extension: %s: %v", e.ID, err) } diff --git a/internal/app/hostcaps.go b/internal/app/hostcaps.go new file mode 100644 index 0000000..05a941f --- /dev/null +++ b/internal/app/hostcaps.go @@ -0,0 +1,116 @@ +package app + +import ( + "context" + "fmt" + + "agent-system-backend/internal/rag" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + core "agent-system-core" +) + +// hostcaps 聚合 owner 作用域的宿主能力（存储、检索管线、LLM 路由）。 +// 同一实现同时服务 hostapi.Server 的 Search/Complete 闭包与进程内 +// inprocHost（EXT-1 批注 1：复用现有 owner 作用域函数，不新造句柄）。 +type hostcaps struct { + st *store.Store + pipe *rag.Pipeline + set *settings.Service +} + +// search owner 作用域混合检索；projectID/topK 为空/0 时用管线默认值。 +func (c hostcaps) search(ctx context.Context, owner string, projectID core.ID, text string, topK int) ([]core.RetrievalHit, error) { + return c.pipe.Retrieve(ctx, core.ID(owner), core.RetrievalQuery{ProjectID: projectID, Text: text, TopK: topK}) +} + +// searchDefault 与 hostapi /host/v1 检索端点同口径（不带 projectID/topK）。 +func (c hostcaps) searchDefault(ctx context.Context, owner, text string) ([]core.RetrievalHit, error) { + return c.search(ctx, owner, "", text, 0) +} + +// complete 经基座 LLM 路由补全（owner 的模型配置生效；未配置报错）。 +func (c hostcaps) complete(ctx context.Context, owner, prompt string) (string, error) { + route := runtime.ResolveChatModel(core.AgentProfile{}, c.set.Effective(owner, "")) + if route.Model == "" || route.Base == "" { + return "", fmt.Errorf("llm route not configured") + } + return completeChat(ctx, route.Base, route.Key, route.Model, prompt) +} + +// inprocHost 是进程内插件命令的 owner 作用域宿主句柄（extension.Host 实现）。 +// 每次分发按请求 Owner 新建；所有能力调用都以该 owner 为第一参数，插件无法越权。 +type inprocHost struct { + caps hostcaps + pluginID string + owner string +} + +func (h inprocHost) PluginID() string { return h.pluginID } +func (h inprocHost) Owner() string { return h.owner } + +func (h inprocHost) KVPut(key, value string) error { + return h.caps.st.PutPluginKV(h.pluginID, h.owner, key, value) +} + +func (h inprocHost) KVGet(key string) (string, bool, error) { + return h.caps.st.GetPluginKV(h.pluginID, h.owner, key) +} + +func (h inprocHost) Audit(action, target, evidence string) error { + return h.caps.st.AppendAuditEvent(store.AuditEvent{ + Owner: h.owner, Actor: "plugin:" + h.pluginID, + Action: action, Target: target, Evidence: evidence, + }) +} + +func (h inprocHost) Complete(ctx context.Context, prompt string) (string, error) { + return h.caps.complete(ctx, h.owner, prompt) +} + +func (h inprocHost) KnowledgeSearch(ctx context.Context, projectID core.ID, text string, topK int) ([]core.RetrievalHit, error) { + return h.caps.search(ctx, h.owner, projectID, text, topK) +} + +// KnowledgeWrite 强制 owner 归属：Owner 改写为本次分发 owner，Source 为空补 +// "plugin:"；embed=false 只写全文行，embed=true 走 IndexChunks 建向量 +//（embedding 失败降级纯全文，不阻断写入）。 +func (h inprocHost) KnowledgeWrite(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error { + if len(docs) == 0 { + return nil + } + src := "plugin:" + h.pluginID + for i := range docs { + docs[i].Owner = core.ID(h.owner) + if docs[i].Source == "" { + docs[i].Source = src + } + } + if !embed { + return h.caps.st.AddKnowledge(docs...) + } + return h.caps.pipe.IndexChunks(ctx, core.ID(h.owner), docs) +} + +// TaskCreate/TaskUpdate 后台任务台账：基座只记进度/状态，执行体由下游自跑（H5）。 +func (h inprocHost) TaskCreate(kind, title, conversationID string) (string, error) { + t, err := h.caps.st.CreateBackgroundTask(h.owner, kind, title, conversationID) + if err != nil { + return "", err + } + return t.ID, nil +} + +func (h inprocHost) TaskUpdate(taskID, status string, progress int, errText string) error { + return h.caps.st.UpdateBackgroundTask(h.owner, taskID, status, progress, errText) +} + +// ArtifactGet/ArtifactContent 工件只读（R1-3 ingest 读原件；写入属完整集）。 +func (h inprocHost) ArtifactGet(artifactID string) (core.Artifact, error) { + return h.caps.st.GetArtifact(h.owner, artifactID) +} + +func (h inprocHost) ArtifactContent(artifactID string) ([]byte, error) { + return h.caps.st.ArtifactContent(h.owner, artifactID) +} diff --git a/internal/app/inproc_host_test.go b/internal/app/inproc_host_test.go new file mode 100644 index 0000000..ea9026a --- /dev/null +++ b/internal/app/inproc_host_test.go @@ -0,0 +1,195 @@ +package app + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" + "time" + + agentcommand "agent-command" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/rag" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +// newInprocTestApp 组合根测试版：走真实的 applyCompiled → cmdbus.Dispatch → +// inprocHost 链路（卡片端口 nil，仅普通危险级命令；embedder 非 nil 时为 hash 档）。 +func newInprocTestApp(t *testing.T, embedder func() string) (*cmdbus.Bus, *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() }) + set := settings.NewService(st) + pipe := rag.NewPipeline(st, set, nil, nil) + if embedder != nil { + pipe.SetEmbedder(rag.SwitchingEmbedder{Provider: embedder}) + } + bus := cmdbus.New(st, nil, func() time.Time { return time.Unix(1700000000, 0).UTC() }) + caps := hostcaps{st: st, pipe: pipe, set: set} + applyCompiled(filepath.Join(t.TempDir(), "skills"), bus, caps) + return bus, st +} + +// TestInprocHostOwnerScope 是 D8/H4 烟测的单元级前置：两个 owner 各分发一次， +// 审计、知识、KV、任务台账必须各落自己 owner 名下（D8 烟测的 PG 版另行）。 +func TestInprocHostOwnerScope(t *testing.T) { + extension.ResetCompiled() + t.Cleanup(extension.ResetCompiled) + extension.RegisterCompiled(extension.InProc{ + ID: "tst", + Register: func(reg extension.Registry) error { + return reg.AddCommandHandler(json.RawMessage(`{"id":"tst.record","title":"记录"}`), + func(ctx context.Context, h extension.Host, req extension.Request) (any, error) { + ref, _ := req.Params["ref"].(string) + if err := h.Audit("tst.record", ref, "{}"); err != nil { + return nil, err + } + if err := h.KnowledgeWrite(ctx, []core.KnowledgeDoc{ + {Title: "分块", Ref: ref, Chunk: string(h.Owner()) + " 的入库内容"}, + }, true); err != nil { + return nil, err + } + if err := h.KVPut("last", string(h.Owner())); err != nil { + return nil, err + } + taskID, err := h.TaskCreate("tst.record", "记录 "+ref, req.ConversationID) + if err != nil { + return nil, err + } + if err := h.TaskUpdate(taskID, "succeeded", 100, ""); err != nil { + return nil, err + } + hits, err := h.KnowledgeSearch(ctx, "", string(h.Owner())+" 的入库内容", 5) + if err != nil { + return nil, err + } + return map[string]any{"owner": h.Owner(), "plugin": h.PluginID(), "hits": len(hits)}, nil + }) + }, + }) + bus, st := newInprocTestApp(t, nil) // embedder nil：embed=true 仍只落全文（IndexChunks 降级路径） + + for _, owner := range []string{"alice", "bob"} { + resp := bus.Dispatch(context.Background(), cmdbus.Request{ + ID: "tst.record", Params: map[string]any{"ref": owner + "-ref"}, Owner: owner, Actor: "ui", Entrance: "chat", + }) + if resp.Receipt.State != agentcommand.StateCommitted { + t.Fatalf("%s dispatch: %+v", owner, resp.Receipt) + } + m, _ := resp.Payload.(map[string]any) + if m["owner"] != owner || m["plugin"] != "tst" || m["hits"] != 1 { + t.Fatalf("%s payload: %+v", owner, resp.Payload) + } + } + + // 审计各归其主，Actor 标注插件身份。 + for _, owner := range []string{"alice", "bob"} { + evs, err := st.ListAuditEvents(owner, 10) + if err != nil { + t.Fatal(err) + } + if len(evs) != 1 || evs[0].Owner != owner || evs[0].Actor != "plugin:tst" || evs[0].Action != "tst.record" { + t.Fatalf("%s audit: %+v", owner, evs) + } + } + + // 知识各归其主：本人可见，他人检索不可见（visibility 过滤）。 + aHits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "alice 的入库内容"}) + if err != nil { + t.Fatal(err) + } + if len(aHits) != 1 || string(aHits[0].Doc.Owner) != "alice" { + t.Fatalf("alice knowledge: %+v", aHits) + } + cross, err := st.SearchKnowledge("bob", core.RetrievalQuery{Text: "alice 的入库内容"}) + if err != nil { + t.Fatal(err) + } + if len(cross) != 0 { + t.Fatalf("owner leak in knowledge: %+v", cross) + } + + // KV 按 pluginID+owner 隔离。 + for _, owner := range []string{"alice", "bob"} { + v, ok, err := st.GetPluginKV("tst", owner, "last") + if err != nil || !ok || v != owner { + t.Fatalf("%s kv: %q %v %v", owner, v, ok, err) + } + if _, ok, _ := st.GetPluginKV("other-plugin", owner, "last"); ok { + t.Fatalf("%s kv leaked across plugins", owner) + } + } + + // 任务台账各归其主且状态已更新。 + for _, owner := range []string{"alice", "bob"} { + tasks, err := st.ListBackgroundTasks(owner) + if err != nil { + t.Fatal(err) + } + if len(tasks) != 1 || tasks[0].Kind != "tst.record" || tasks[0].Status != "succeeded" || tasks[0].Owner != owner { + t.Fatalf("%s tasks: %+v", owner, tasks) + } + } +} + +// TestInprocHostKnowledgeWriteEmbedHash：provider=hash 档下 KnowledgeWrite(embed=true) +// 写出向量行，且 Owner/Source 由基座强制改写。 +func TestInprocHostKnowledgeWriteEmbedHash(t *testing.T) { + extension.ResetCompiled() + t.Cleanup(extension.ResetCompiled) + extension.RegisterCompiled(extension.InProc{ + ID: "tst", + Register: func(reg extension.Registry) error { + return reg.AddCommandHandler(json.RawMessage(`{"id":"tst.embed","title":"向量"}`), + func(ctx context.Context, h extension.Host, req extension.Request) (any, error) { + text, _ := req.Params["text"].(string) + // 故意不带 Owner/Source：验证基座强制改写。 + return "ok", h.KnowledgeWrite(ctx, []core.KnowledgeDoc{{Title: "向量", Ref: "vec-1", Chunk: text}}, true) + }) + }, + }) + bus, st := newInprocTestApp(t, func() string { return "hash" }) + + resp := bus.Dispatch(context.Background(), cmdbus.Request{ + ID: "tst.embed", Params: map[string]any{"text": "Hash 向量内容"}, Owner: "alice", Entrance: "api", + }) + if resp.Receipt.State != agentcommand.StateCommitted { + t.Fatalf("dispatch: %+v", resp.Receipt) + } + + // Owner 强制 alice、Source 强制补 plugin:tst。 + hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "向量内容"}) + if err != nil { + t.Fatal(err) + } + if len(hits) != 1 || string(hits[0].Doc.Owner) != "alice" || hits[0].Doc.Source != "plugin:tst" { + t.Fatalf("doc attribution: %+v", hits) + } + + // hash 档向量行存在且可被同文本向量召回。 + want, err := rag.HashEmbedder{}.Embed(context.Background(), []string{"Hash 向量内容"}) + if err != nil { + t.Fatal(err) + } + vhits, err := st.SearchKnowledgeVectors(context.Background(), "alice", want[0], 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(vhits) != 1 { + t.Fatalf("vector rows: %+v", vhits) + } + // bob 用同一向量检索不可见。 + bhits, err := st.SearchKnowledgeVectors(context.Background(), "bob", want[0], 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(bhits) != 0 { + t.Fatalf("vector owner leak: %+v", bhits) + } +} diff --git a/internal/cmdbus/cmdbus.go b/internal/cmdbus/cmdbus.go index e867905..17e6512 100644 --- a/internal/cmdbus/cmdbus.go +++ b/internal/cmdbus/cmdbus.go @@ -140,6 +140,18 @@ func (b *Bus) RegisterPluginCommands(overlay []json.RawMessage) error { return nil } +// RegisterPluginCommand 注册进程内插件命令（EXT-1）：登记规格并绑定宿主处理器。 +// 仅登记规格的声明式路径仍走 RegisterPluginCommands（无 handler，分发即 rejected）。 +func (b *Bus) RegisterPluginCommand(spec agentcommand.CommandSpec, h Handler) error { + if err := b.reg.Register(spec, agentcommand.SourcePlugin); err != nil { + return fmt.Errorf("cmdbus: register %s: %w", spec.ID, err) + } + if h != nil { + b.handlers[spec.ID] = h + } + return nil +} + // ResolveSlash `/` 入口解析：返回命令 id 与原始字段（参数由宿主处理器按命令语义解析）。 func (b *Bus) ResolveSlash(cmd string) (string, bool) { id, ok := b.slash[strings.ToLower(strings.TrimSpace(cmd))] diff --git a/internal/cmdbus/cmdbus_test.go b/internal/cmdbus/cmdbus_test.go index 0a32a56..4b78adc 100644 --- a/internal/cmdbus/cmdbus_test.go +++ b/internal/cmdbus/cmdbus_test.go @@ -211,3 +211,72 @@ func TestValidateParamsScalars(t *testing.T) { t.Fatalf("nil schema must pass: %v", err) } } + +// ---- EXT-1：进程内插件命令处理器 ---- + +func TestRegisterPluginCommandDispatch(t *testing.T) { + a := &memAudit{} + b := newTestBus(a, &memCards{}) + err := b.RegisterPluginCommand(agentcommand.CommandSpec{ID: "rdm.ingest", Title: "入库"}, func(ctx context.Context, req Request) (any, error) { + return map[string]any{"echo": req.Params["ref"], "owner": req.Owner}, nil + }) + if err != nil { + t.Fatal(err) + } + resp := b.Dispatch(context.Background(), Request{ID: "rdm.ingest", Params: map[string]any{"ref": "doc-1"}, Owner: "alice"}) + if resp.Receipt.State != agentcommand.StateCommitted { + t.Fatalf("dispatch: %+v", resp.Receipt) + } + m, _ := resp.Payload.(map[string]any) + if m["echo"] != "doc-1" || m["owner"] != "alice" { + t.Fatalf("payload: %+v", resp.Payload) + } + if len(a.logs) != 1 || a.logs[0].Source != agentcommand.SourcePlugin { + t.Fatalf("audit: %+v", a.logs) + } + // 重复注册同一 ID：fail-closed。 + if err := b.RegisterPluginCommand(agentcommand.CommandSpec{ID: "rdm.ingest", Title: "重复"}, nil); err == nil { + t.Fatal("duplicate plugin command must be rejected") + } +} + +func TestPluginCommandWithoutHandlerRejected(t *testing.T) { + b := newTestBus(&memAudit{}, &memCards{}) + // 声明式路径只登记规格（manifest provides.commands）。 + if err := b.RegisterPluginCommands([]json.RawMessage{json.RawMessage(`{"id":"plug.only","title":"仅规格"}`)}); err != nil { + t.Fatal(err) + } + resp := b.Dispatch(context.Background(), Request{ID: "plug.only"}) + if resp.Receipt.State != agentcommand.StateRejected || !strings.Contains(resp.Receipt.Reason, "no host handler bound") { + t.Fatalf("unbound: %+v", resp.Receipt) + } +} + +func TestHighDangerPluginCommandConfirmsFirst(t *testing.T) { + a := &memAudit{} + cards := &memCards{} + b := newTestBus(a, cards) + executed := 0 + err := b.RegisterPluginCommand(agentcommand.CommandSpec{ + ID: "rdm.purge", Title: "清库", Danger: agentcommand.DangerHigh, + }, func(ctx context.Context, req Request) (any, error) { + executed++ + return "purged", nil + }) + if err != nil { + t.Fatal(err) + } + // 未确认：先出内联确认卡，处理器不得执行。 + resp := b.Dispatch(context.Background(), Request{ID: "rdm.purge", ConversationID: "cnv_9", Owner: "alice"}) + if resp.Receipt.State != agentcommand.StateAccepted || resp.Card == nil || resp.Card.Kind != core.CardKindConfirm { + t.Fatalf("pre-confirm: %+v card=%+v", resp.Receipt, resp.Card) + } + if executed != 0 { + t.Fatal("handler must not run before confirm") + } + // 确认后执行 committed。 + resp2 := b.Confirm("alice", "card_x1", "confirm") + if resp2.Receipt.State != agentcommand.StateCommitted || resp2.Payload != "purged" || executed != 1 { + t.Fatalf("confirm: %+v executed=%d", resp2.Receipt, executed) + } +} diff --git a/internal/rag/hybrid.go b/internal/rag/hybrid.go index f683186..2f8a502 100644 --- a/internal/rag/hybrid.go +++ b/internal/rag/hybrid.go @@ -80,32 +80,50 @@ func (p *Pipeline) IndexDocument(owner, projectID core.ID, source, title, ref, p for i, c := range chunks { docs[i] = core.KnowledgeDoc{Owner: owner, ProjectID: projectID, Source: source, Title: title, Ref: ref, Chunk: c.Text} } + return p.IndexChunks(context.Background(), owner, docs) +} + +// IndexChunks 写入已构造好的知识分块并建向量（EXT-1 KnowledgeWrite embed=true 的路径）。 +// AddKnowledge 原地回填 docs[i].ID；embedding 失败降级为纯全文（记日志，不阻断写入）。 +func (p *Pipeline) IndexChunks(ctx context.Context, owner core.ID, docs []core.KnowledgeDoc) error { + if len(docs) == 0 { + return nil + } if err := p.st.AddKnowledge(docs...); err != nil { return err } if p.embedder == nil { return nil } - texts := make([]string, len(chunks)) - for i, c := range chunks { - texts[i] = c.Text + texts := make([]string, len(docs)) + for i, d := range docs { + texts[i] = d.Chunk } - vecs, err := p.embedder.Embed(context.Background(), texts) + vecs, err := p.embedder.Embed(ctx, texts) if err != nil { - log.Printf("[rag] embedding 降级为纯全文（ref=%s）: %v", ref, err) + log.Printf("[rag] embedding 降级为纯全文（ref=%s）: %v", firstRef(docs), err) return nil } for i, v := range vecs { if len(v) == 0 { continue } - if err := p.st.UpsertKnowledgeVector(context.Background(), owner, docs[i].ID, ref, v); err != nil { + if err := p.st.UpsertKnowledgeVector(ctx, owner, docs[i].ID, docs[i].Ref, v); err != nil { return err } } return nil } +func firstRef(docs []core.KnowledgeDoc) string { + for _, d := range docs { + if d.Ref != "" { + return d.Ref + } + } + return "" +} + // Retrieve 统一检索（core.Retriever）。 func (p *Pipeline) Retrieve(ctx context.Context, owner core.ID, q core.RetrievalQuery) ([]core.RetrievalHit, error) { topK := q.TopK diff --git a/internal/rag/rag_test.go b/internal/rag/rag_test.go index 9f38de7..8260039 100644 --- a/internal/rag/rag_test.go +++ b/internal/rag/rag_test.go @@ -228,3 +228,68 @@ func TestVectorDimMismatchSkipped(t *testing.T) { t.Fatalf("dim mismatch must skip, got %+v", hits) } } + +// ---- EXT-1：IndexChunks（KnowledgeWrite embed=true 的底层路径） ---- + +func TestIndexChunksWritesFulltextAndVectors(t *testing.T) { + p, st := newTestPipeline(t, fakeEmbedder{}, nil) + docs := []core.KnowledgeDoc{ + {Owner: "alice", Source: "plugin:rdm", Title: "规格", Ref: "spec-1", Chunk: "Alpha 内容甲"}, + {Owner: "alice", Source: "plugin:rdm", Title: "规格", Ref: "spec-1", Chunk: "beta 内容乙"}, + } + if err := p.IndexChunks(context.Background(), "alice", docs); err != nil { + t.Fatal(err) + } + // ID 已被 AddKnowledge 回填，向量按回填 ID 关联。 + for _, d := range docs { + if d.ID == "" { + t.Fatal("doc ID not backfilled") + } + } + // 全文行落库。 + hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "内容甲"}) + if err != nil { + t.Fatal(err) + } + if len(hits) == 0 { + t.Fatal("fulltext row missing") + } + // 向量行落库（fakeEmbedder：非 A 前缀 → [0,1]）。 + vhits, err := st.SearchKnowledgeVectors(context.Background(), "alice", []float32{0, 1}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(vhits) == 0 { + t.Fatal("vector rows missing") + } + // owner 作用域：bob 检索不到 alice 的向量。 + bhits, err := st.SearchKnowledgeVectors(context.Background(), "bob", []float32{0, 1}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(bhits) != 0 { + t.Fatalf("owner leak: %+v", bhits) + } +} + +func TestIndexChunksEmbedFailureDegrades(t *testing.T) { + p, st := newTestPipeline(t, fakeEmbedder{fail: true}, nil) + docs := []core.KnowledgeDoc{{Owner: "alice", Source: "plugin:rdm", Ref: "spec-2", Chunk: "降级内容"}} + if err := p.IndexChunks(context.Background(), "alice", docs); err != nil { + t.Fatalf("embed failure must degrade to fulltext, got %v", err) + } + hits, err := st.SearchKnowledge("alice", core.RetrievalQuery{Text: "降级"}) + if err != nil { + t.Fatal(err) + } + if len(hits) == 0 { + t.Fatal("fulltext row missing after degrade") + } + vhits, err := st.SearchKnowledgeVectors(context.Background(), "alice", []float32{0, 1}, 5, core.RetrievalQuery{}) + if err != nil { + t.Fatal(err) + } + if len(vhits) != 0 { + t.Fatalf("no vectors expected after degrade, got %+v", vhits) + } +} diff --git a/pkg/extension/extension.go b/pkg/extension/extension.go index 6bf70f2..835590c 100644 --- a/pkg/extension/extension.go +++ b/pkg/extension/extension.go @@ -18,6 +18,9 @@ var ErrNotLinked = errors.New("extension: runner not linked; blank-import agent- type Registry interface { AddSkill(name, markdown string) error AddCommand(spec json.RawMessage) error + // AddCommandHandler 注册插件命令规格并绑定进程内处理器（EXT-1）： + // 基座在分发时按请求 Owner 构造 Host 句柄调用 h。 + AddCommandHandler(spec json.RawMessage, h Handler) error Note(kind, name string, payload json.RawMessage) error } diff --git a/pkg/extension/host.go b/pkg/extension/host.go new file mode 100644 index 0000000..4397a88 --- /dev/null +++ b/pkg/extension/host.go @@ -0,0 +1,55 @@ +package extension + +import ( + "context" + + core "agent-system-core" +) + +// Request 是进程内命令处理器收到的一次分发请求（cmdbus.Request 的子集）。 +type Request struct { + ID string + Params map[string]any + Entrance string // button | card | slash | chat | api（归一） + Actor string + Owner string + ConversationID string +} + +// Handler 是下游注册的进程内命令处理器：每次分发由基座构造 owner 作用域的 +// Host 句柄传入，处理器经句柄访问基座能力，不得自行越权持有 owner 凭证。 +type Handler func(ctx context.Context, h Host, req Request) (any, error) + +// Host 是基座在分发时构造的 owner 作用域宿主句柄（EXT-1 最小集）。 +// 所有能力的可见范围都以本次分发的 Owner 为准；artifacts 只读 +//（写入属完整集，另行评估）。实现由基座 internal 提供。 +type Host interface { + // PluginID 返回注册处理器的插件标识（形如 rdm）。 + PluginID() string + // Owner 返回本次分发的作用域 owner。 + Owner() string + + // KVPut/KVGet 插件 KV（按 pluginID+owner 隔离）。 + KVPut(key, value string) error + KVGet(key string) (value string, ok bool, err error) + + // Audit 写一条 owner 名下的审计事件。 + Audit(action, target, evidence string) error + + // Complete 经基座 LLM 路由补全（owner 的模型配置生效）。 + Complete(ctx context.Context, prompt string) (string, error) + + // KnowledgeSearch 混合检索（owner 可见范围）。 + KnowledgeSearch(ctx context.Context, projectID core.ID, text string, topK int) ([]core.RetrievalHit, error) + // KnowledgeWrite 写知识分块；embed=true 时同步建向量（失败降级纯全文）。 + // docs 各项的 Owner 由基座强制改写为本次 owner；Source 为空补 "plugin:"。 + KnowledgeWrite(ctx context.Context, docs []core.KnowledgeDoc, embed bool) error + + // TaskCreate/TaskUpdate 后台任务台账（进度/状态记账，执行体由下游自跑）。 + TaskCreate(kind, title, conversationID string) (taskID string, err error) + TaskUpdate(taskID, status string, progress int, errText string) error + + // ArtifactGet/ArtifactContent 工件只读（R1-3 ingest 读原件）。 + ArtifactGet(artifactID string) (core.Artifact, error) + ArtifactContent(artifactID string) ([]byte, error) +} diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index 4381290..79f0298 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -1,5 +1,7 @@ Compiled ErrNotLinked +Handler +Host HostClient InProc Main @@ -7,6 +9,7 @@ NewHostClient Options RegisterCompiled Registry +Request ResetCompiled SetRunner SurfaceVersion -- 2.54.0