From ce5bff9275015306fca2edbae804cc5b1810f5c0 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 6 Sep 2026 10:03:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(sdk):=20=E5=A4=9A=E6=A8=A1=E6=80=81?= =?UTF-8?q?=E8=B4=AF=E9=80=9A=E6=8F=92=E4=BB=B6=E8=BE=B9=E7=95=8C=E2=80=94?= =?UTF-8?q?=E2=80=94=E5=AA=92=E4=BD=93=E5=AD=97=E6=AE=B5=E3=80=81=E5=AA=92?= =?UTF-8?q?=E4=BD=93=E6=B3=A8=E5=85=A5=E6=8E=A5=E5=8F=A3=E4=B8=8E=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 记忆系统在核心 1.1.0 支持了二进制多媒体节点，但那条链路只对**内核自己**开放： 插件把 Triple / Doc 交进来，媒体一律无处安放，且**不报错**。本版补上公开接口 侧缺失的表达能力。 ## 一、类型与接口（全部新增，无签名变更） - `Triple` += `SentenceText`、`MediaDigests` - `Doc` += `MediaDigests`、`Attachments`；新增 `MediaAttachment` - `TextEvent` += `Attachments` - `DocMemoryAPI` += `InsertWithMedia` - `IOInjector` += `InjectInputMedia` / `InjectInputMediaSync` / `InjectInterruptMedia` - `PluginSDK` 补上一直缺失的 `SetToolBlocks` 包装（接口里有、便捷方法里没有， 插件只能自己去拿 injector） `MediaAttachment` 一个类型服务两个方向：给 `Data`+`MIME` 是新内容（内核按字节 去重），只给 `Digest` 是引用已有内容。读路径**只回元数据不回字节**——一次检索 可能命中几十份媒体，把字节全塞回来会撑爆跨进程消息。 媒体注入为什么不能搭 `SetToolBlocks` 的车：那个方法只在工具处理函数内部可用， 且媒体要等**下一条** tool message 才到模型手上。插件主动发起一轮带媒体的对话、 以及中断注入，需要各自的签名，且媒体在**本轮**就随消息发出。 `Triple.MediaDigests` 非空而 `SentenceText` 为空时，内核会用媒体标记本身充当句子 ——媒体引用挂在句子上，没有句子就无处挂起。插件只需填 digest，标记由内核拼： 要求调用方知道格式，等于让一个拼写错误静默切断引用绑定而全链路无人报错。 ## 二、修掉两处并发竞态 `sdk/stress_test.go` 的 `-race` 实测报 11 处 DATA RACE，收敛到两个字段： 1. **`PluginSDK` 的 API 字段无锁**。写方是内核（加载/重载插件时依次注入 injector、memory、doc、llm…），读方是插件在 `Start()` 里起的后台 goroutine ——轮询、监听、定时器都要拿 injector 往管道注消息。生产表现是插件重载瞬间 偶发崩溃：读到半个接口值就 nil 解引用。 2. **`autoRestart` 标志无锁**。`SetAutoRestart` 的文档用法本身就是「外部连接建好 后再决定能否自动重启」，而连接建立通常在后台 goroutine；内核 registry 在另一个 goroutine 读 `AutoRestart()` 决定崩溃后重启策略。这对读写天然跨 goroutine。 加 `apiMu sync.RWMutex`。关键约定写进注释：**只在持锁期间取字段值，取完立刻 释放再调用**。持锁调用会把 `InjectInputSync`（阻塞到 agent 回复，可达数分钟） 与 `SetIOInjector` 串到一起，让插件重载卡死。 ## 三、压测（sdk/stress_test.go，13 例） SDK 是被多个 goroutine 同时使用的共享对象，单线程单测全绿不代表并发路径成立。 断言的是不变量而非吞吐： - 媒体注入高并发不丢不串——每次调用带唯一 tag，逐条校验文本与图片 URL 配对。 「不串」是重点：若实现里出现任何共享中间状态（把 blocks 暂存到字段再读出）， 高并发下会出现 A 的文本配 B 的图，而两者单独看都「成功」了； - injector 热替换（含替换成 nil，即内核卸载 API 的真实状态）； - stop / onRemove handler 恰好一次——契约是「执行后清空，幂等」，执行两次的后果 从重复写文件到 close 已关闭 channel 直接 panic； - `StageContext` 并发读改写无 lost update（媒体链路让 Extra 成为新热点， 而 map 并发写在 Go 里是直接 fatal，recover 接不住）； - `OwnTools` scope 不跨插件泄漏； - 媒体类型 JSON 往返字节级一致（9 种长度，含 0/1/2/3 与 base64 分组边界） ——`[]byte` 在 JSON 里是 base64，往返不一致意味着图片静默损坏， 要到 CAS 校验 digest 时才发现，那时已无从追查； - `omitempty` 真的生效（读路径不能出现 `"data"` 键）； - nil 依赖全部静默降级不 panic。 ## 四、工具链同步 - `proc_main.go.tmpl`：`procIO` 三个媒体方法、`procDocMemory.InsertWithMedia`。 模板不跟上的后果是**每个外部插件都编不过**（接口未实现），是硬失败； - `proc_runtime_test.go`：方法清单补 `io.injectMedia*` 与 `doc.insertWithMedia`。 漏接线时插件调 `InjectInputMedia` 会静默无效果——模板不发这个 RPC，内核也就 收不到，两边都不报错； - `yaegi/mocksdk`：与公开 SDK 对齐。它此前漂移严重且**没有任何代码对着它编译**， 所以漂移不会被编译器抓到：`Triple` 用的是 `Predicate`，而公开 SDK 一直叫 `Relation` —— 插件在 yaegi 调试期写 `Relation:` 报未知字段，写 `Predicate:` 则 编成 plugin.bin 时报错，两边都不对。 - README 中英双语补媒体接口文档与用法示例。 ## 兼容性 存量插件不需要改一行也不需要重编：新增方法由**插件调用、内核实现**，不调就不 受影响。17 个 example 插件源码零改动通过类型检查；用 SDK 0.9.2 编的旧 plugin.bin 在新内核上直接建链通过（握手校验的是 ProtocolVersion=1，不是 SDK 版本）。 媒体接口需要核心 1.1.1+（更早的核心没有对应 RPC，调用返回 unknown method）。 `CoreVersion` 保持 1.0.0：它是「SDK 能在其上运行」的下限，媒体是可选能力。 --- README.md | 80 +++ README_EN.md | 86 +++ meta/meta.go | 30 +- sdk/memory.go | 64 +- sdk/plugin.go | 257 +++++-- sdk/stress_test.go | 725 ++++++++++++++++++++ tools/plugindev/proc_runtime_test.go | 5 + tools/plugindev/templates/proc_main.go.tmpl | 48 ++ tools/plugindev/yaegi/mocksdk/plugin.go | 141 +++- 9 files changed, 1339 insertions(+), 97 deletions(-) create mode 100644 sdk/stress_test.go diff --git a/README.md b/README.md index 4568b88..20d86d3 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ type Plugin interface { | 设置 | `Settings()` | 访问设置 API | | 事件 | `Events()` | 访问事件订阅器（外部插件仅订阅） | | 注入 | `InjectText(source, channel, text)` / `InjectInterruptText(source, channel, text)` / `InjectTextNoMemory(source, channel, text)` | 向管道注入文本 | +| 多模态注入 | `InjectInputMedia(source, channel, text, blocks)` / `InjectInputMediaSync(...)` / `InjectInterruptMedia(...)` | 注入带图片/音频的输入（1.1.0 新增） | | 自动重启 | `SetAutoRestart(enabled)` / `AutoRestart()` | 控制崩溃自动重启 | ### 阶段钩子 @@ -106,6 +107,30 @@ type 枚举值： | `InjectInterruptText(source, channel, text)` | 注入中断文本，打断当前处理，路由到指定通道 | | `InjectTextNoMemory(source, channel, text)` | 注入文本，不记入内存，路由到指定通道 | +### 多模态注入（1.1.0 新增） + +| 方法 | 说明 | +|------|------| +| `InjectInputMedia(source, channel, text, blocks)` | 注入带媒体的输入，异步 | +| `InjectInputMediaSync(source, channel, text, blocks)` | 注入带媒体的输入并同步等待回复文本 | +| `InjectInterruptMedia(source, channel, text, blocks)` | 注入带媒体的中断，可抢占当前处理 | + +`blocks` 是 `[]sdk.ContentBlock`，与 `SetToolBlocks` 用同一类型： + +```go +s.InjectInputMedia("myplugin", "webui", "帮我看看这张图", []sdk.ContentBlock{{ + Type: "image_url", + ImageURL: &sdk.ImageURL{URL: "data:image/png;base64," + b64, Detail: "auto"}, +}}) +``` + +与 `SetToolBlocks` 的区别：`SetToolBlocks` 只能在工具处理函数内部调用，媒体要等到 +下一条 tool message 才到模型手上；这三个方法是插件**主动发起一轮带媒体的对话**， +媒体在本轮就随消息发给模型，并自动落进媒体存储、挂上媒体记忆引用。 + +媒体块里的 `data:` URL 会被内核落盘去重；`http(s)` URL 只透传给模型，不入库 +（入库需要内核发起网络请求，涉及超时、鉴权与 SSRF）。 + `source` 标识来源，`channel` 指定目标输出通道。 ### Triple 扩展字段 @@ -115,6 +140,61 @@ Triple 数据结构新增字段： - `Confidence` — 置信度（0.0~1.0） - `SubjectType` — 主体类型 - `ObjectType` — 客体类型 +- `SentenceText` — 原始句子文本（1.1.0 新增），写入 `sentences` 表；媒体引用挂在句子上 +- `MediaDigests` — 关联的媒体 digest 列表（1.1.0 新增） + +### 记忆里的媒体（1.1.0 新增） + +媒体在纯文本记忆里以**标记**形式存在，格式 `[ <短digest>] <描述>`： + +``` +[image/png a1b2c3d4e5f6] 一张紫蓝红三色带图 +``` + +描述文本是持久的语义记忆（检索靠它），digest 是回到字节的钥匙（反查靠它）。 +标记由内核生成，插件不必自己拼——**填 digest 就够**。 + +#### 图记忆 + +```go +s.Memory().Commit([]sdk.Triple{{ + Subject: "配色图", Relation: "包含", Object: "三色带", + MediaDigests: []string{"a1b2c3d4e5f6"}, // 短 digest 即可，内核补全 +}}) +``` + +没给 `SentenceText` 时内核会用标记本身充当句子——媒体必须有句子落点， +否则引用无从挂起。 + +#### 知识库 + +```go +s.DocMemory().InsertWithMedia(&sdk.Doc{ + Title: "带图笔记", + Content: "正文", +}, []sdk.MediaAttachment{ + {MIME: "image/png", Data: pngBytes, Name: "chart.png"}, // 新内容，落盘去重 + {Digest: "a1b2c3d4e5f6"}, // 引用已有内容 +}) +``` + +`Insert` 保持原签名不变，正文里已有的标记同样会被挂成文档级引用。 +`Query` 返回的 `Doc` 带 `MediaDigests` 与 `Attachments`（mime + 描述， +**不含字节**——一次检索可能命中几十份媒体）。删除文档时引用自动释放。 + +#### 文本记忆 + +```go +s.TextMemory().Append(sdk.TextEvent{ + Role: "user", Content: "看这张图", + Attachments: []sdk.MediaAttachment{{MIME: "image/png", Data: pngBytes}}, +}) +``` + +`RecentEvents` 读回时正文里的标记会被反解成 `Attachments`。 + +媒体存储可在内核侧关闭（`core.memory.media.enabled=false`），此时以上接口 +全部退化为纯文本行为：不报错、不 panic，与本特性上线前一致。 ### ToolDef 字段说明 diff --git a/README_EN.md b/README_EN.md index c20dce6..9bb56ff 100644 --- a/README_EN.md +++ b/README_EN.md @@ -36,6 +36,7 @@ The SDK instance injected via `Start(sdk *PluginSDK)` provides: | Settings | `Settings()` | Access settings API | | Events | `Events()` | Access event subscriber (subscribe-only for external plugins) | | Inject | `InjectText(source, channel, text)` / `InjectInterruptText(source, channel, text)` / `InjectTextNoMemory(source, channel, text)` | Inject text into the agent pipeline | +| Media inject | `InjectInputMedia(source, channel, text, blocks)` / `InjectInputMediaSync(...)` / `InjectInterruptMedia(...)` | Inject input carrying images/audio (added in 1.1.0) | | Auto-Restart | `SetAutoRestart(enabled)` / `AutoRestart()` | Control automatic restart on crash | ### Stage Hooks @@ -106,6 +107,32 @@ Type enum values: | `InjectInterruptText(source, channel, text)` | Inject interrupt text, interrupt current processing, route to specified channel | | `InjectTextNoMemory(source, channel, text)` | Inject text without memory recording, route to specified channel | +### Multimodal Injection (added in 1.1.0) + +| Method | Description | +|--------|-------------| +| `InjectInputMedia(source, channel, text, blocks)` | Inject media-bearing input, asynchronous | +| `InjectInputMediaSync(source, channel, text, blocks)` | Inject media-bearing input and wait for the reply text | +| `InjectInterruptMedia(source, channel, text, blocks)` | Inject a media-bearing interrupt that can preempt current processing | + +`blocks` is `[]sdk.ContentBlock`, the same type `SetToolBlocks` takes: + +```go +s.InjectInputMedia("myplugin", "webui", "take a look at this", []sdk.ContentBlock{{ + Type: "image_url", + ImageURL: &sdk.ImageURL{URL: "data:image/png;base64," + b64, Detail: "auto"}, +}}) +``` + +How this differs from `SetToolBlocks`: that one is only callable inside a tool handler and +its media reaches the model with the *next* tool message. These three let a plugin +**initiate a turn that carries media** — the media goes out with this turn's message and is +automatically stored in the media store with a memory reference attached. + +`data:` URLs in the blocks are stored and deduplicated by the kernel; `http(s)` URLs are +passed to the model only and never stored (storing them would require the kernel to make +network requests, bringing timeouts, auth and SSRF into scope). + `source` identifies the origin, `channel` specifies the target output channel. ### Triple Extended Fields @@ -115,6 +142,65 @@ The Triple data structure includes additional fields: - `Confidence` — confidence score (0.0–1.0) - `SubjectType` — subject type - `ObjectType` — object type +- `SentenceText` — the original sentence (added in 1.1.0), written to the `sentences` table; media references hang off the sentence +- `MediaDigests` — associated media digests (added in 1.1.0) + +### Media in Memory (added in 1.1.0) + +Inside plain-text memory, media is represented as a **marker** of the form +`[ ] `: + +``` +[image/png a1b2c3d4e5f6] a purple-blue-red three-band chart +``` + +The description is the durable semantic memory (retrieval uses it); the digest is the key +back to the bytes (reverse lookup uses it). Markers are generated by the kernel — a plugin +never has to assemble one, it just **supplies the digest**. + +#### Graph memory + +```go +s.Memory().Commit([]sdk.Triple{{ + Subject: "palette", Relation: "contains", Object: "three-band", + MediaDigests: []string{"a1b2c3d4e5f6"}, // short digest is fine, the kernel resolves it +}}) +``` + +With no `SentenceText`, the kernel uses the marker itself as the sentence — media must have +a sentence to hang off, otherwise the reference has nowhere to attach. + +#### Knowledge base + +```go +s.DocMemory().InsertWithMedia(&sdk.Doc{ + Title: "illustrated note", + Content: "body", +}, []sdk.MediaAttachment{ + {MIME: "image/png", Data: pngBytes, Name: "chart.png"}, // new content, stored and deduped + {Digest: "a1b2c3d4e5f6"}, // reference existing content +}) +``` + +`Insert` keeps its original signature; markers already present in the body are bound as +document-level references too. `Query` fills `MediaDigests` and `Attachments` (mime plus +description, **no bytes** — one query can match dozens of media items). Removing a document +releases its references. + +#### Text memory + +```go +s.TextMemory().Append(sdk.TextEvent{ + Role: "user", Content: "look at this", + Attachments: []sdk.MediaAttachment{{MIME: "image/png", Data: pngBytes}}, +}) +``` + +`RecentEvents` decodes markers in the body back into `Attachments`. + +The media store can be disabled kernel-side (`core.memory.media.enabled=false`); all of the +above then degrades to plain-text behaviour — no errors, no panics, identical to how it +behaved before this feature shipped. ### ToolDef Field Reference diff --git a/meta/meta.go b/meta/meta.go index 1e0619c..598cd70 100644 --- a/meta/meta.go +++ b/meta/meta.go @@ -6,9 +6,30 @@ var ( // Version 是 HomeAgent SDK 版本号。 // 通过 `-ldflags="-X gitcode.com/JianFeeeee/homeagent-sdk/meta.Version=vX.Y.Z"` 注入。 // + // 版本号语义：**SDK 版本跟随核心的中版本，patch 位恒为 .0**。 + // 整条核心 1.1.x 线（1.1.0、1.1.1、1.1.7…）共用 SDK 1.1.0； + // 只有核心进入 1.2.0 这种中版本跃迁时 SDK 才升到 1.2.0。 + // 这样插件开发者只需关心「我在为哪个中版本写插件」， + // 不必跟着核心的每个 bugfix 换 SDK 依赖（见 核心仓 docs/git-branching.md §七）。 + // // 1.0.0：插件运行模型从 C ABI 动态库改为子进程 + 共享内存。 - // 公开 SDK 接口（sdk/ 目录）**零改动**——插件业务代码不需要改一行， - // 但产物形态变了（plugin.so → plugin.bin），必须用新版 plugindev 重编。 + // 公开 SDK 接口零改动，但产物形态变了（plugin.so → plugin.bin）。 + // 1.1.0：多模态贯通插件边界。**全部是新增，无签名变更**： + // - Triple.SentenceText / Triple.MediaDigests + // - Doc.MediaDigests / Doc.Attachments、MediaAttachment + // - TextEvent.Attachments + // - DocMemoryAPI.InsertWithMedia + // - IOInjector 的 InjectInputMedia / InjectInputMediaSync / + // InjectInterruptMedia；PluginSDK 补上缺失的 SetToolBlocks 包装 + // 同版修掉两处并发竞态（sdk/stress_test.go 的 -race 实证，不是理论风险）： + // PluginSDK 的 API 字段与 autoRestart 标志此前无锁，而写方 + // （内核注入 API、插件 SetAutoRestart）与读方（插件后台 goroutine + // 注入、内核 registry 读 AutoRestart）天然跨 goroutine。 + // 存量插件不需要改一行也不需要重编：新增方法由**插件调用、内核实现**， + // 不调就不受影响。想用新字段的插件重编即可。 + // + // ❗main 分支上此值是**下一个未发布中版本**；已发布的值看对应的 + // release/vX.Y.x 分支与 tag（见 核心仓 docs/git-branching.md §2.1 与 §七.1）。 Version = "1.0.0" // Commit 是构建时的 Git commit hash。 @@ -27,6 +48,11 @@ var ( // // 1.0.0 是硬下限而非建议值：0.9.x 内核只会 dlopen `.so`， // 本版工具链产出的 `plugin.bin` 在旧内核上根本不会被识别。 + // + // ⚠️ 1.1.0 新增的媒体接口需要核心 **1.1.1+**（更早的核心没有 + // doc.insertWithMedia / io.injectMedia* 这些 RPC，调用会返回 unknown method）。 + // 这里仍写 1.0.0，因为它是「SDK 能在其上运行」的下限； + // 媒体接口是可选能力，不用就不受影响。 CoreVersion = "1.0.0" ) diff --git a/sdk/memory.go b/sdk/memory.go index 215b313..a90785f 100644 --- a/sdk/memory.go +++ b/sdk/memory.go @@ -25,13 +25,18 @@ type Relation struct { } // Triple represents a subject-relation-object triple for the knowledge graph. +// +// SentenceText 是这条三元组的原句，会写进 sentences 表；媒体引用挂在句子上， +// 所以 MediaDigests 非空时内核会保证句子存在（不给就自动合成一句）。 type Triple struct { - Subject string `json:"subject"` - Relation string `json:"relation"` - Object string `json:"object"` - Confidence float64 `json:"confidence,omitempty"` - SubjectType string `json:"subject_type,omitempty"` - ObjectType string `json:"object_type,omitempty"` + Subject string `json:"subject"` + Relation string `json:"relation"` + Object string `json:"object"` + Confidence float64 `json:"confidence,omitempty"` + SubjectType string `json:"subject_type,omitempty"` + ObjectType string `json:"object_type,omitempty"` + SentenceText string `json:"sentence_text,omitempty"` + MediaDigests []string `json:"media_digests,omitempty"` } // TextMemoryAPI provides access to chronological text event storage. @@ -40,27 +45,52 @@ type TextMemoryAPI interface { } // TextEvent represents a single text memory event. +// MediaAttachment 描述一份与记忆关联的媒体。 +// +// 两个方向共用一个类型： +// - 写入（InsertWithMedia）：给 Data + MIME 就是新内容；只给 Digest 则是引用已有内容。 +// - 读出（Query）：内核只填 Digest/MIME/Description，**不回 Data**—— +// 一次检索可能命中几十张图，把字节全塞回插件会把 ABI 消息撑爆。 +// 需要字节时拿 Digest 单独取。 +type MediaAttachment struct { + Digest string `json:"digest,omitempty"` + MIME string `json:"mime,omitempty"` + Data []byte `json:"data,omitempty"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` +} + type TextEvent struct { - Role string `json:"role"` - Content string `json:"content"` - Timestamp int64 `json:"timestamp"` - Channel string `json:"channel,omitempty"` + Role string `json:"role"` + Content string `json:"content"` + Timestamp int64 `json:"timestamp"` + Channel string `json:"channel,omitempty"` + Attachments []MediaAttachment `json:"attachments,omitempty"` } // DocMemoryAPI provides access to the document vector store. type DocMemoryAPI interface { Query(text string, topK int) []*Doc Insert(doc *Doc) error + // InsertWithMedia 写入文档并关联媒体。attachments 里带 Data 的会落进 + // 内容寻址存储（相同字节只存一份），只带 Digest 的直接引用已有内容。 + // 插件无需自己拼标记：内核会把 `[mime <短 digest>] <描述>` 补进 Content， + // 让向量检索和后续蒸馏都能看到这份媒体。 + InsertWithMedia(doc *Doc, attachments []MediaAttachment) error Remove(id string) Stats() map[string]interface{} } // Doc represents a document in the document store. +// +// MediaDigests / Attachments 在 Query 返回时由内核填充（仅元数据，不带字节）。 type Doc struct { - ID string `json:"id"` - Title string `json:"title"` - Content string `json:"content"` - Score float64 `json:"score,omitempty"` + ID string `json:"id"` + Title string `json:"title"` + Content string `json:"content"` + Score float64 `json:"score,omitempty"` + MediaDigests []string `json:"media_digests,omitempty"` + Attachments []MediaAttachment `json:"attachments,omitempty"` } // SocialAPI provides read-only access to the social graph (person profiles and relationships). @@ -75,9 +105,9 @@ type SocialAPI interface { // PersonProfile represents a person's complete profile (traits + social relations). type PersonProfile struct { - Name string `json:"name"` - Traits map[string]string `json:"traits,omitempty"` - Relations []SocialRelation `json:"relations,omitempty"` + Name string `json:"name"` + Traits map[string]string `json:"traits,omitempty"` + Relations []SocialRelation `json:"relations,omitempty"` } // SocialRelation represents a social relationship between two persons. diff --git a/sdk/plugin.go b/sdk/plugin.go index 0d7221a..ebd234d 100644 --- a/sdk/plugin.go +++ b/sdk/plugin.go @@ -61,14 +61,18 @@ type StageContext struct { Memory []MemItem NoMemory bool Extra map[string]interface{} - Errors []string // 阶段处理过程中的错误信息 + Errors []string // 阶段处理过程中的错误信息 } -func (c *StageContext) RLock() { c.mu.RLock() } -func (c *StageContext) RUnlock() { c.mu.RUnlock() } -func (c *StageContext) Lock() { c.mu.Lock() } -func (c *StageContext) Unlock() { c.mu.Unlock() } -func (c *StageContext) IsResponded() bool { c.mu.RLock(); defer c.mu.RUnlock(); return c.Response != nil } +func (c *StageContext) RLock() { c.mu.RLock() } +func (c *StageContext) RUnlock() { c.mu.RUnlock() } +func (c *StageContext) Lock() { c.mu.Lock() } +func (c *StageContext) Unlock() { c.mu.Unlock() } +func (c *StageContext) IsResponded() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.Response != nil +} // MemItem represents a memory item in stage context. type MemItem struct { @@ -100,8 +104,8 @@ type ToolDef struct { Plugin string `json:"plugin,omitempty"` Description string `json:"description"` Parameters map[string]interface{} `json:"parameters"` - NoMemory bool `json:"no_memory,omitempty"` // 此工具输出不参与记忆计算，但原文保留 - Cleaner func(string) string `json:"-"` // 计算层过滤函数，不改原文；仅在向量化/jieba/蒸馏时调用 + NoMemory bool `json:"no_memory,omitempty"` // 此工具输出不参与记忆计算，但原文保留 + Cleaner func(string) string `json:"-"` // 计算层过滤函数，不改原文；仅在向量化/jieba/蒸馏时调用 } // IOInjector provides methods for injecting input and interrupts into the agent pipeline. @@ -117,6 +121,9 @@ type IOInjector interface { // SetToolBlocks 插件工具注入多模态内容块（image_url/audio_url），内核在下一条 // tool message 的 content 数组里带上这些块，让模型在后续轮次看到图/听到音频。 SetToolBlocks(blocks []ContentBlock) + InjectInputMedia(source, channel, text string, blocks []ContentBlock) + InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string + InjectInterruptMedia(source, channel, text string, blocks []ContentBlock) } // EventType identifies the kind of system event. @@ -221,6 +228,24 @@ type PluginSDK struct { events EventSubscriber plgMgr PluginMgrAPI + // apiMu 保护上面这些由内核注入的 API 字段，以及 autoRestart。 + // + // 这些字段的写方与读方天然跨 goroutine： + // - 写方是内核（加载/重载插件时注入 API）与插件自己（SetAutoRestart）； + // - 读方是插件在 Start() 里起的后台 goroutine（轮询、监听、定时器 + // 都要拿 injector 往管道里注消息），以及内核 registry —— 它在 + // 另一个 goroutine 读 AutoRestart() 决定崩溃后是否重启。 + // SetAutoRestart 的文档用法本身就是「连接建立后再决定能否自动重启」， + // 而连接建立通常发生在后台 goroutine 里，于是这对读写必然并发。 + // + // sdk/stress_test.go 的 -race 实测确认这是真竞态，不是理论风险。 + // 未加锁时的生产表现是偶发 nil 解引用崩溃（读到半个接口值）。 + // + // 约定：只在持锁期间取字段值，取完立刻释放再调用。 + // 持锁调用会把 InjectInputSync 这类阻塞到 agent 回复（可达数分钟）的 + // 方法与 SetIOInjector 串到一起，让插件重载卡死。 + apiMu sync.RWMutex + autoRestart bool stopMu sync.Mutex @@ -247,28 +272,57 @@ func New(name string, sett SettingsAPI, regTool ToolRegistrar, regStage StageReg func (s *PluginSDK) PluginName() string { return s.name } // Settings returns the settings API for reading/writing plugin configuration. +// sett 在 New 时一次性写入且无 setter，故不需要加锁。 func (s *PluginSDK) Settings() SettingsAPI { return s.sett } // Memory returns the graph memory API (may be nil if not available). -func (s *PluginSDK) Memory() MemoryAPI { return s.mem } +func (s *PluginSDK) Memory() MemoryAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.mem +} // TextMemory returns the text memory API (may be nil if not available). -func (s *PluginSDK) TextMemory() TextMemoryAPI { return s.textMem } +func (s *PluginSDK) TextMemory() TextMemoryAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.textMem +} // DocMemory returns the document memory API (may be nil if not available). -func (s *PluginSDK) DocMemory() DocMemoryAPI { return s.docMem } +func (s *PluginSDK) DocMemory() DocMemoryAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.docMem +} // Knowledge returns the knowledge store API (may be nil if not available). -func (s *PluginSDK) Knowledge() KnowledgeAPI { return s.know } +func (s *PluginSDK) Knowledge() KnowledgeAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.know +} // LLM returns the LLM provider API (may be nil if not available). -func (s *PluginSDK) LLM() LLMAPI { return s.llm } +func (s *PluginSDK) LLM() LLMAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.llm +} // Social returns the social graph API (may be nil if not available). -func (s *PluginSDK) Social() SocialAPI { return s.social } +func (s *PluginSDK) Social() SocialAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.social +} // Events returns the event subscriber for listening to kernel events (may be nil if not available). -func (s *PluginSDK) Events() EventSubscriber { return s.events } +func (s *PluginSDK) Events() EventSubscriber { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.events +} // RegisterTool registers a tool that the LLM can call. func (s *PluginSDK) RegisterTool(name string, def ToolDef, handler ToolHandler) error { @@ -282,8 +336,9 @@ func (s *PluginSDK) RegisterTool(name string, def ToolDef, handler ToolHandler) } // RegisterStage registers a handler for a pipeline stage. -// scope: StageScopeGlobal (default) — receives all stage events. -// StageScopeOwnTools — only before_toolcall/after_toolcall for this plugin's tools. +// +// scope: StageScopeGlobal (default) — receives all stage events. +// StageScopeOwnTools — only before_toolcall/after_toolcall for this plugin's tools. func (s *PluginSDK) RegisterStage(stage Stage, handler StageHandler, scope ...StageScope) { if s.regStage == nil { return @@ -333,8 +388,11 @@ func (s *PluginSDK) RegisterPluginAPI(name string) error { // def: 通道在记忆计算层的行为（NoMemory/Cleaner） // handler: receives args map with keys: payload (string), type (string), meta (string|optional) func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, def ChannelDef, handler ToolHandler) error { - if s.regOutput != nil { - return s.regOutput(name, caps, desc, def, handler) + s.apiMu.RLock() + reg := s.regOutput + s.apiMu.RUnlock() + if reg != nil { + return reg(name, caps, desc, def, handler) } return nil } @@ -343,57 +401,126 @@ func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, de // def.NoMemory: 此通道输入不参与记忆计算 // def.Cleaner: 计算层对输入文本清洗后（不改原文）再向量化/提关键词 func (s *PluginSDK) RegisterInputChannel(name string, def ChannelDef) error { - if s.regInput != nil { - return s.regInput(name, def) + s.apiMu.RLock() + reg := s.regInput + s.apiMu.RUnlock() + if reg != nil { + return reg(name, def) } return nil } +// 以下 setter 由内核在启动/重载时调用，与插件后台 goroutine 的读并发，故加锁。 + // SetOutputChannelRegistrar sets the output channel registrar (called by the core at startup). -func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) { s.regOutput = r } +func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) { + s.apiMu.Lock() + s.regOutput = r + s.apiMu.Unlock() +} // SetInputChannelRegistrar sets the input channel registrar (called by the core at startup). -func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) { s.regInput = r } +func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) { + s.apiMu.Lock() + s.regInput = r + s.apiMu.Unlock() +} // SetIOInjector sets the IO injector (called by the core at startup). -func (s *PluginSDK) SetIOInjector(io IOInjector) { s.io = io } +func (s *PluginSDK) SetIOInjector(io IOInjector) { + s.apiMu.Lock() + s.io = io + s.apiMu.Unlock() +} // SetMemoryAPI sets the memory API (called by the core at startup). -func (s *PluginSDK) SetMemoryAPI(mem MemoryAPI) { s.mem = mem } -func (s *PluginSDK) SetTextMemoryAPI(tm TextMemoryAPI) { s.textMem = tm } -func (s *PluginSDK) SetDocMemoryAPI(dm DocMemoryAPI) { s.docMem = dm } -func (s *PluginSDK) SetKnowledgeAPI(kn KnowledgeAPI) { s.know = kn } -func (s *PluginSDK) SetLLMAPI(llm LLMAPI) { s.llm = llm } -func (s *PluginSDK) SetSocialAPI(social SocialAPI) { s.social = social } -func (s *PluginSDK) SetEventSubscriber(es EventSubscriber) { s.events = es } +func (s *PluginSDK) SetMemoryAPI(mem MemoryAPI) { + s.apiMu.Lock() + s.mem = mem + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetTextMemoryAPI(tm TextMemoryAPI) { + s.apiMu.Lock() + s.textMem = tm + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetDocMemoryAPI(dm DocMemoryAPI) { + s.apiMu.Lock() + s.docMem = dm + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetKnowledgeAPI(kn KnowledgeAPI) { + s.apiMu.Lock() + s.know = kn + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetLLMAPI(llm LLMAPI) { + s.apiMu.Lock() + s.llm = llm + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetSocialAPI(social SocialAPI) { + s.apiMu.Lock() + s.social = social + s.apiMu.Unlock() +} + +func (s *PluginSDK) SetEventSubscriber(es EventSubscriber) { + s.apiMu.Lock() + s.events = es + s.apiMu.Unlock() +} // SetPluginMgrAPI sets the plugin manager API (called by the bridge at startup). -func (s *PluginSDK) SetPluginMgrAPI(pm PluginMgrAPI) { s.plgMgr = pm } +func (s *PluginSDK) SetPluginMgrAPI(pm PluginMgrAPI) { + s.apiMu.Lock() + s.plgMgr = pm + s.apiMu.Unlock() +} // PluginMgr returns the plugin manager API (ReloadOne / ReloadPlugins / list). // May be nil if the host did not wire it. -func (s *PluginSDK) PluginMgr() PluginMgrAPI { return s.plgMgr } +func (s *PluginSDK) PluginMgr() PluginMgrAPI { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.plgMgr +} // ---- IO Convenience Methods ---- +// injector 取当前 injector 的快照。 +// +// 取完即释放锁再调用：InjectInputSync 会阻塞到 agent 回复（可达数分钟）， +// 若持锁调用，插件重载时的 SetIOInjector 会一起卡住。 +func (s *PluginSDK) injector() IOInjector { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.io +} + // InjectInterruptText injects a text interrupt that can preempt current LLM processing. func (s *PluginSDK) InjectInterruptText(source, channel, text string) { - if s.io != nil { - s.io.InjectInterruptText(source, channel, text) + if io := s.injector(); io != nil { + io.InjectInterruptText(source, channel, text) } } // InjectText injects a text message into the agent pipeline. func (s *PluginSDK) InjectText(source, channel, text string) { - if s.io != nil { - s.io.InjectText(source, channel, text) + if io := s.injector(); io != nil { + io.InjectText(source, channel, text) } } // InjectTextNoMemory injects a text message without generating memory. func (s *PluginSDK) InjectTextNoMemory(source, channel, text string) { - if s.io != nil { - s.io.InjectTextNoMemory(source, channel, text) + if io := s.injector(); io != nil { + io.InjectTextNoMemory(source, channel, text) } } @@ -401,18 +528,62 @@ func (s *PluginSDK) InjectTextNoMemory(source, channel, text string) { // returning the reply text (empty string if none). Replies must be dispatched back // to the source channel by the caller. func (s *PluginSDK) InjectInputSync(source, channel, text string) string { - if s.io == nil { + io := s.injector() + if io == nil { return "" } - return s.io.InjectInputSync(source, channel, text) + return io.InjectInputSync(source, channel, text) +} + +// InjectInputMedia 注入带媒体内容块（image_url/audio_url）的输入。 +// blocks 会落进媒体存储被记忆引用捕获，同时作为当前轮 content 数组 +// 发给 LLM，让模型在「本轮」就看到图/听到音频——区别于 SetToolBlocks +// 的「下一轮 tool message」语义。 +func (s *PluginSDK) InjectInputMedia(source, channel, text string, blocks []ContentBlock) { + if io := s.injector(); io != nil { + io.InjectInputMedia(source, channel, text, blocks) + } +} + +// InjectInputMediaSync 注入带媒体内容块的输入并同步等待 agent 回复。 +func (s *PluginSDK) InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string { + io := s.injector() + if io == nil { + return "" + } + return io.InjectInputMediaSync(source, channel, text, blocks) +} + +// InjectInterruptMedia 注入带媒体内容块的中断，可抢占当前 LLM 处理。 +// blocks 随中断消息一起发给模型。 +func (s *PluginSDK) InjectInterruptMedia(source, channel, text string, blocks []ContentBlock) { + if io := s.injector(); io != nil { + io.InjectInterruptMedia(source, channel, text, blocks) + } +} + +// SetToolBlocks 在工具处理函数内注入多模态内容块，内核在下一条 tool message +// 的 content 数组里带上它们。需要「本轮就让模型看到」时用 InjectInputMedia。 +func (s *PluginSDK) SetToolBlocks(blocks []ContentBlock) { + if io := s.injector(); io != nil { + io.SetToolBlocks(blocks) + } } // SetAutoRestart 设置插件是否允许内核自动重启（崩溃后自动重载）。 // 默认 true。如果插件有无法恢复的状态（如外部连接），应设为 false。 -func (s *PluginSDK) SetAutoRestart(enabled bool) { s.autoRestart = enabled } +func (s *PluginSDK) SetAutoRestart(enabled bool) { + s.apiMu.Lock() + s.autoRestart = enabled + s.apiMu.Unlock() +} // AutoRestart 返回插件是否允许自动重启。 -func (s *PluginSDK) AutoRestart() bool { return s.autoRestart } +func (s *PluginSDK) AutoRestart() bool { + s.apiMu.RLock() + defer s.apiMu.RUnlock() + return s.autoRestart +} // RegisterStopHandler 注册插件停止阶段的清理回调。 // 注册的 handler 会在插件 Stop() 之前按"后注册先执行"的顺序调用， diff --git a/sdk/stress_test.go b/sdk/stress_test.go new file mode 100644 index 0000000..c088cdc --- /dev/null +++ b/sdk/stress_test.go @@ -0,0 +1,725 @@ +package sdk + +import ( + "encoding/json" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" +) + +// SDK 公开接口的并发压力测试（1.1.0 媒体接口上线后新增）。 +// +// 为什么这一层需要压测：SDK 是**被多个 goroutine 同时使用的共享对象**。 +// 一个插件的典型形态是 Start() 里起若干后台 goroutine（轮询、监听、定时器）， +// 它们各自持同一个 *PluginSDK 往里注入消息；内核侧同时还有 stage 扇出、 +// 工具调用、以及读 AutoRestart() 决定崩溃后是否重启。 +// 单线程单测全绿不代表这些并发路径成立。 +// +// 关注点不是吞吐数字，而是不变量： +// 1. 注入调用不丢、不串（媒体块必须与文本配对，不能张冠李戴） +// 2. 状态字段的读写不产生数据竞争（-race 下必须干净） +// 3. handler 注册/执行在并发下"恰好一次" +// 4. 跨进程 JSON 序列化对新媒体类型必须字节级往返一致 +// +// 媒体接口尤其需要 3 与 4：媒体块要经 JSON 过子进程边界， +// 而 []byte 在 JSON 里是 base64，往返不一致的后果是图片静默损坏。 + +// ---------- 测试替身 ---------- + +// recordingInjector 记录每一次注入调用，用于验证"不丢不串"。 +type recordingInjector struct { + mu sync.Mutex + calls []injectCall + + // 计数用原子量：并发路径上只增不减，可在不持锁时安全读。 + nText, nMedia, nInterrupt, nSync atomic.Int64 +} + +type injectCall struct { + kind string // text / media / interruptMedia / sync ... + source string + channel string + text string + blocks []ContentBlock +} + +func (r *recordingInjector) record(c injectCall) { + r.mu.Lock() + r.calls = append(r.calls, c) + r.mu.Unlock() +} + +func (r *recordingInjector) InjectInterruptText(s, c, t string) { + r.nInterrupt.Add(1) + r.record(injectCall{kind: "interruptText", source: s, channel: c, text: t}) +} + +func (r *recordingInjector) InjectText(s, c, t string) { + r.nText.Add(1) + r.record(injectCall{kind: "text", source: s, channel: c, text: t}) +} + +func (r *recordingInjector) InjectTextNoMemory(s, c, t string) { + r.nText.Add(1) + r.record(injectCall{kind: "textNoMem", source: s, channel: c, text: t}) +} + +func (r *recordingInjector) InjectInputSync(s, c, t string) string { + r.nSync.Add(1) + r.record(injectCall{kind: "sync", source: s, channel: c, text: t}) + return "reply:" + t +} + +func (r *recordingInjector) SetToolBlocks(blocks []ContentBlock) { + r.record(injectCall{kind: "toolBlocks", blocks: blocks}) +} + +func (r *recordingInjector) InjectInputMedia(s, c, t string, b []ContentBlock) { + r.nMedia.Add(1) + r.record(injectCall{kind: "media", source: s, channel: c, text: t, blocks: b}) +} + +func (r *recordingInjector) InjectInputMediaSync(s, c, t string, b []ContentBlock) string { + r.nMedia.Add(1) + r.nSync.Add(1) + r.record(injectCall{kind: "mediaSync", source: s, channel: c, text: t, blocks: b}) + return "reply:" + t +} + +func (r *recordingInjector) InjectInterruptMedia(s, c, t string, b []ContentBlock) { + r.nMedia.Add(1) + r.record(injectCall{kind: "interruptMedia", source: s, channel: c, text: t, blocks: b}) +} + +func (r *recordingInjector) snapshot() []injectCall { + r.mu.Lock() + defer r.mu.Unlock() + return append([]injectCall{}, r.calls...) +} + +var _ IOInjector = (*recordingInjector)(nil) + +// imageBlock 构造一个带可识别 URL 的图片块。 +func imageBlock(tag string) ContentBlock { + return ContentBlock{ + Type: "image_url", + ImageURL: &ImageURL{URL: "data:image/png;base64," + tag, Detail: "auto"}, + } +} + +// ---------- 1. 媒体注入并发不丢不串 ---------- + +// 三个媒体注入方法在高并发下必须：调用数精确、且每次调用的 text 与 blocks 配对不错。 +// +// "不串"是这里的关键断言。注入是插件里最容易被后台 goroutine 并发调用的入口， +// 若实现里出现任何共享中间状态（比如把 blocks 暂存到 SDK 字段再读出）， +// 高并发下就会出现 A 的文本配上 B 的图——而两者单独看都"成功"了，不报错。 +func TestStress_MediaInjectionConcurrentNoCrossTalk(t *testing.T) { + const workers, perWorker = 32, 200 + + inj := &recordingInjector{} + s := &PluginSDK{name: "stress"} + s.SetIOInjector(inj) + + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < perWorker; i++ { + // tag 唯一标识这次调用，文本与图片 URL 里都带上它。 + tag := fmt.Sprintf("w%d-i%d", w, i) + switch i % 3 { + case 0: + s.InjectInputMedia("src", "ch", tag, []ContentBlock{imageBlock(tag)}) + case 1: + if got := s.InjectInputMediaSync("src", "ch", tag, []ContentBlock{imageBlock(tag)}); got != "reply:"+tag { + t.Errorf("同步注入回复错位: got %q want %q", got, "reply:"+tag) + } + default: + s.InjectInterruptMedia("src", "ch", tag, []ContentBlock{imageBlock(tag)}) + } + } + }(w) + } + wg.Wait() + + total := int64(workers * perWorker) + if got := inj.nMedia.Load(); got != total { + t.Fatalf("媒体注入调用数 = %d，期望 %d（有调用丢失）", got, total) + } + + // 逐条校验文本与媒体块配对：URL 必须含该次调用自己的 tag。 + seen := map[string]bool{} + for _, c := range inj.snapshot() { + if len(c.blocks) == 0 { + continue + } + if c.blocks[0].ImageURL == nil { + t.Fatalf("媒体块 ImageURL 丢失: %+v", c.blocks[0]) + } + if !strings.HasSuffix(c.blocks[0].ImageURL.URL, c.text) { + t.Fatalf("文本与媒体块错位: text=%q url=%q", c.text, c.blocks[0].ImageURL.URL) + } + if seen[c.text] { + t.Fatalf("同一次调用被记录两次: %s", c.text) + } + seen[c.text] = true + } + if len(seen) != int(total) { + t.Fatalf("去重后调用数 = %d，期望 %d", len(seen), total) + } +} + +// ---------- 2. 注入期间热替换 injector ---------- + +// 内核在插件运行期间可能重新注入 API（重载、恢复、子进程重连握手）。 +// 此时插件的后台 goroutine 仍在注入。这条路径若无同步就是对 s.io 的数据竞争， +// 在 -race 下会被抓出；生产表现是偶发 nil 解引用崩溃。 +func TestStress_InjectorSwapDuringInjection(t *testing.T) { + s := &PluginSDK{name: "stress"} + s.SetIOInjector(&recordingInjector{}) + + stop := make(chan struct{}) + var injectors, swapper sync.WaitGroup + + // 注入方：持续打直到 stop + for w := 0; w < 8; w++ { + injectors.Add(1) + go func() { + defer injectors.Done() + for { + select { + case <-stop: + return + default: + s.InjectInputMedia("src", "ch", "x", []ContentBlock{imageBlock("x")}) + s.InjectText("src", "ch", "y") + } + } + }() + } + + // 替换方：反复换 injector（含换成 nil——内核卸载 API 时的真实状态） + swapper.Add(1) + go func() { + defer swapper.Done() + for i := 0; i < 500; i++ { + if i%7 == 0 { + s.SetIOInjector(nil) + } else { + s.SetIOInjector(&recordingInjector{}) + } + } + }() + + // 先等替换跑完，再告知注入方退出。 + // 顺序写反了就是死锁：注入方只依 close(stop) 退出。 + swapper.Wait() + close(stop) + injectors.Wait() + // 断言就是「没崩、-race 没报」。nil injector 时必须静默跳过而非 panic。 +} + +// ---------- 3. autoRestart 标志的并发读写 ---------- + +// SetAutoRestart 的文档用途是"插件有无法恢复的状态（如外部连接）时设为 false"—— +// 而连接建立本身通常是异步的，所以这个写入天然发生在后台 goroutine。 +// 内核侧 registry 在另一个 goroutine 读 AutoRestart() 决定崩溃后是否重启。 +// 这是一对跨 goroutine 的读写，必须同步。 +func TestStress_AutoRestartFlagConcurrent(t *testing.T) { + s := &PluginSDK{name: "stress", autoRestart: true} + + var wg sync.WaitGroup + for w := 0; w < 16; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < 500; i++ { + s.SetAutoRestart(i%2 == 0) + } + }(w) + } + // 读方模拟内核 registry + for r := 0; r < 8; r++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 500; i++ { + _ = s.AutoRestart() + } + }() + } + wg.Wait() +} + +// ---------- 4. stop / onRemove handler 的"恰好一次" ---------- + +// RunStopHandlers 的契约是"执行后清空，幂等"。内核在停止插件时可能并发触发 +// （超时强杀与正常 Stop 竞争），handler 里往往是关连接、落盘—— +// 执行两次的后果从"重复写文件"到"close 已关闭的 channel 直接 panic"。 +func TestStress_StopHandlersExactlyOnce(t *testing.T) { + const n = 300 + s := &PluginSDK{name: "stress"} + + var counters [n]atomic.Int64 + for i := 0; i < n; i++ { + i := i + s.RegisterStopHandler(func() { counters[i].Add(1) }) + } + + var wg sync.WaitGroup + for w := 0; w < 16; w++ { + wg.Add(1) + go func() { + defer wg.Done() + s.RunStopHandlers() + }() + } + wg.Wait() + + for i := 0; i < n; i++ { + if got := counters[i].Load(); got != 1 { + t.Fatalf("handler %d 执行 %d 次，期望恰好 1 次", i, got) + } + } +} + +// 注册与执行并发：已注册的 handler 一次都不能多跑，未跑到的也不能被丢。 +// 断言用"每个 handler 的执行次数 <= 1"而非"总数相等"—— +// 与 RunStopHandlers 竞争的注册可能落在快照之后，那属于合法的未执行。 +func TestStress_StopHandlersRegisterWhileRunning(t *testing.T) { + s := &PluginSDK{name: "stress"} + const n = 500 + var counters [n]atomic.Int64 + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < n; i++ { + i := i + s.RegisterStopHandler(func() { counters[i].Add(1) }) + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 50; i++ { + s.RunStopHandlers() + } + }() + wg.Wait() + s.RunStopHandlers() // 收尾：把剩下的都跑掉 + + for i := 0; i < n; i++ { + if got := counters[i].Load(); got > 1 { + t.Fatalf("handler %d 被执行 %d 次（重复执行）", i, got) + } + } +} + +func TestStress_OnRemoveHandlersExactlyOnce(t *testing.T) { + const n = 200 + s := &PluginSDK{name: "stress"} + + var counters [n]atomic.Int64 + for i := 0; i < n; i++ { + i := i + s.RegisterOnRemoveHandler(func() { counters[i].Add(1) }) + } + + var wg sync.WaitGroup + for w := 0; w < 12; w++ { + wg.Add(1) + go func() { + defer wg.Done() + s.RunOnRemoveHandlers() + }() + } + wg.Wait() + + for i := 0; i < n; i++ { + if got := counters[i].Load(); got != 1 { + t.Fatalf("onRemove handler %d 执行 %d 次，期望恰好 1 次", i, got) + } + } +} + +// ---------- 5. StageContext 并发读改写 ---------- + +// StageContext 是全部 stage handler 共享的可变状态，字段全导出、靠调用方自觉 +// 持 Lock/RLock。媒体链路让 Extra 成为新热点（media_blocks 挂在这里）， +// 而 map 的并发写在 Go 里是直接 fatal，recover 都接不住。 +// +// 这条测试锁定的不变量：按约定持锁的并发读改写不丢更新、不 fatal。 +func TestStress_StageContextConcurrentExtraAndFinalText(t *testing.T) { + ctx := &StageContext{Extra: map[string]interface{}{}} + + const workers, rounds = 16, 200 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < rounds; i++ { + // 写：模拟插件往 Extra 塞媒体块并追加文本（读-改-写） + ctx.Lock() + ctx.Extra[fmt.Sprintf("k%d-%d", w, i)] = []ContentBlock{imageBlock("x")} + ctx.FinalText += "." + ctx.Unlock() + + // 读：模拟另一个 handler 检查是否已被响应 + _ = ctx.IsResponded() + ctx.RLock() + _ = len(ctx.Extra) + ctx.RUnlock() + } + }(w) + } + wg.Wait() + + ctx.RLock() + defer ctx.RUnlock() + if len(ctx.Extra) != workers*rounds { + t.Fatalf("Extra 键数 = %d，期望 %d（出现 lost update）", len(ctx.Extra), workers*rounds) + } + if len(ctx.FinalText) != workers*rounds { + t.Fatalf("FinalText 长度 = %d，期望 %d（出现 lost update）", len(ctx.FinalText), workers*rounds) + } +} + +// ---------- 6. OwnTools scope 包装器的并发正确性 ---------- + +// StageScopeOwnTools 的包装闭环里要读 ctx.ToolCalls 判断归属。 +// 并发下若判断与执行之间状态被改写，就会出现"别人的工具触发了我的 handler"—— +// 后果是插件对不属于自己的工具结果动手，且没有任何错误。 +func TestStress_OwnToolsScopeNoCrossPluginLeak(t *testing.T) { + var registered StageHandler + s := &PluginSDK{ + name: "mine", + regStage: func(stage Stage, h StageHandler) { registered = h }, + } + + var fired atomic.Int64 + s.RegisterStage(StageBeforeToolcall, func(ctx *StageContext) error { + fired.Add(1) + ctx.RLock() + defer ctx.RUnlock() + // 触发了就必须确实是自己的工具 + if len(ctx.ToolCalls) == 0 || ctx.ToolCalls[0].Plugin != "mine" { + t.Errorf("handler 被别的插件的工具触发: %+v", ctx.ToolCalls) + } + return nil + }, StageScopeOwnTools) + + if registered == nil { + t.Fatal("handler 未注册") + } + + const workers, rounds = 16, 100 + var wg sync.WaitGroup + var mineCount atomic.Int64 + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < rounds; i++ { + // 每个 goroutine 用自己的 ctx——真实内核里 stage 扇出共享同一个 + // ctx，但那部分的并发由内核 host 仲裁；这里验证包装器本身。 + owner := "other" + if (w+i)%2 == 0 { + owner = "mine" + mineCount.Add(1) + } + ctx := &StageContext{Extra: map[string]interface{}{}} + ctx.ToolCalls = []ToolCall{{Plugin: owner, Name: "t"}} + if err := registered(ctx); err != nil { + t.Errorf("handler 返回错误: %v", err) + } + } + }(w) + } + wg.Wait() + + if got, want := fired.Load(), mineCount.Load(); got != want { + t.Fatalf("handler 触发 %d 次，期望 %d 次（漏触发或跨插件触发）", got, want) + } +} + +// ---------- 7. 媒体类型的 JSON 往返（跨进程边界的真实形态） ---------- + +// 媒体块与附件要经 JSON 过子进程边界。[]byte 在 JSON 里是 base64， +// 往返不一致的后果是图片字节静默损坏——落进 CAS 后 digest 校验才会发现， +// 而那时已经无从追查是谁改坏的。 +func TestStress_MediaTypesJSONRoundTripAtScale(t *testing.T) { + // 覆盖真实会遇到的边界：空、单字节、含 0x00、全 0xFF、超过 base64 分组边界的长度 + sizes := []int{0, 1, 2, 3, 255, 256, 1023, 4096, 65537} + for _, n := range sizes { + data := make([]byte, n) + for i := range data { + data[i] = byte(i * 7 % 256) + } + att := MediaAttachment{ + Digest: strings.Repeat("a", 64), + MIME: "image/png", + Data: data, + Name: "图片-名字 with space & 符号.png", + Description: "一张紫蓝红三色带图，含 emoji 🎨 与换行\n第二行", + } + b, err := json.Marshal(att) + if err != nil { + t.Fatalf("size=%d marshal: %v", n, err) + } + var back MediaAttachment + if err := json.Unmarshal(b, &back); err != nil { + t.Fatalf("size=%d unmarshal: %v", n, err) + } + if len(back.Data) != n { + t.Fatalf("size=%d 往返后长度 = %d", n, len(back.Data)) + } + for i := range data { + if back.Data[i] != data[i] { + t.Fatalf("size=%d 第 %d 字节损坏: %02x != %02x", n, i, back.Data[i], data[i]) + } + } + if back.Name != att.Name || back.Description != att.Description || back.MIME != att.MIME || back.Digest != att.Digest { + t.Fatalf("size=%d 元数据往返不一致: %+v", n, back) + } + } +} + +// omitempty 必须真的生效：读路径上内核不回 Data，若序列化仍产出 "data":null +// 之类的键，跨进程消息会凭空变大，且插件侧无法区分"没有字节"与"空字节"。 +func TestStress_MediaTypesOmitEmpty(t *testing.T) { + cases := []struct { + name string + v interface{} + absent []string + present []string + }{ + { + name: "Triple 无媒体", + v: Triple{Subject: "甲方", Relation: "签署", Object: "合同"}, + absent: []string{"media_digests", "sentence_text", "confidence", "subject_type", "object_type"}, + present: []string{"subject", "relation", "object"}, + }, + { + name: "Triple 带媒体", + v: Triple{Subject: "甲方", Relation: "包含", Object: "图", MediaDigests: []string{"abc12345"}, SentenceText: "句子"}, + absent: []string{"confidence"}, + present: []string{"media_digests", "sentence_text"}, + }, + { + name: "Doc 读路径无字节", + v: Doc{ID: "d1", Title: "标题", Content: "正文", Attachments: []MediaAttachment{{Digest: "abc12345", MIME: "image/png"}}}, + absent: []string{"\"data\"", "media_digests", "score"}, + present: []string{"attachments", "digest", "mime"}, + }, + { + name: "TextEvent 无附件", + v: TextEvent{Role: "user", Content: "hi"}, + absent: []string{"attachments", "channel"}, + present: []string{"role", "content"}, + }, + { + name: "ContentBlock 纯文本", + v: ContentBlock{Type: "text", Text: "hi"}, + absent: []string{"image_url", "audio_url"}, + present: []string{"type", "text"}, + }, + { + name: "ContentBlock 图片", + v: imageBlock("AAA"), + absent: []string{"audio_url", "\"text\""}, + present: []string{"image_url", "detail"}, + }, + } + for _, c := range cases { + b, err := json.Marshal(c.v) + if err != nil { + t.Fatalf("%s marshal: %v", c.name, err) + } + s := string(b) + for _, k := range c.absent { + if strings.Contains(s, k) { + t.Errorf("%s: 不该出现的键 %s —— %s", c.name, k, s) + } + } + for _, k := range c.present { + if !strings.Contains(s, k) { + t.Errorf("%s: 缺少键 %s —— %s", c.name, k, s) + } + } + } +} + +// 媒体块在并发序列化下必须各自独立：ImageURL/AudioURL 是指针， +// 若某处复用同一个指针再改写，序列化结果会互相污染。 +func TestStress_ContentBlockConcurrentMarshal(t *testing.T) { + const workers, rounds = 16, 300 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < rounds; i++ { + tag := fmt.Sprintf("w%d-i%d", w, i) + blocks := []ContentBlock{ + {Type: "text", Text: tag}, + imageBlock(tag), + {Type: "audio_url", AudioURL: &AudioURL{URL: "data:audio/wav;base64," + tag}}, + } + b, err := json.Marshal(blocks) + if err != nil { + t.Errorf("marshal: %v", err) + return + } + var back []ContentBlock + if err := json.Unmarshal(b, &back); err != nil { + t.Errorf("unmarshal: %v", err) + return + } + if len(back) != 3 { + t.Errorf("块数 = %d", len(back)) + return + } + if back[0].ImageURL != nil || back[0].AudioURL != nil { + t.Errorf("文本块被填了媒体指针: %+v", back[0]) + } + if back[1].ImageURL == nil || !strings.HasSuffix(back[1].ImageURL.URL, tag) { + t.Errorf("图片块 URL 错位: %+v", back[1].ImageURL) + } + if back[1].AudioURL != nil { + t.Errorf("图片块被填了音频指针") + } + if back[2].AudioURL == nil || !strings.HasSuffix(back[2].AudioURL.URL, tag) { + t.Errorf("音频块 URL 错位: %+v", back[2].AudioURL) + } + } + }(w) + } + wg.Wait() +} + +// ---------- 8. 注册面的并发 ---------- + +// 插件在 Start() 里起多个 goroutine 分别注册工具是常见写法。 +// def.Plugin 的默认填充若不是每次调用独立的，就会出现工具归属错乱—— +// 表现是 OwnTools scope 失效、WebUI 里工具挂在别的插件名下。 +func TestStress_RegisterToolConcurrentPluginDefaulting(t *testing.T) { + var mu sync.Mutex + got := map[string]string{} // toolName -> def.Plugin + + s := &PluginSDK{ + name: "mine", + regTool: func(name string, def ToolDef, h ToolHandler) error { + mu.Lock() + got[name] = def.Plugin + mu.Unlock() + return nil + }, + } + + const workers, perWorker = 16, 100 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < perWorker; i++ { + name := fmt.Sprintf("tool_w%d_i%d", w, i) + def := ToolDef{Description: "d", Parameters: map[string]interface{}{}} + // 一半显式指定归属，一半靠 SDK 填默认值 + if i%2 == 0 { + def.Plugin = "explicit" + } + if err := s.RegisterTool(name, def, func(map[string]interface{}) (interface{}, error) { + return nil, nil + }); err != nil { + t.Errorf("RegisterTool: %v", err) + } + } + }(w) + } + wg.Wait() + + if len(got) != workers*perWorker { + t.Fatalf("注册工具数 = %d，期望 %d", len(got), workers*perWorker) + } + for name, owner := range got { + want := "mine" + if isEvenSuffix(name) { + want = "explicit" + } + if owner != want { + t.Fatalf("工具 %s 归属 = %q，期望 %q", name, owner, want) + } + } +} + +// isEvenSuffix 判断 tool_wX_iY 里的 Y 是否为偶数。 +func isEvenSuffix(name string) bool { + idx := strings.LastIndex(name, "_i") + if idx < 0 { + return false + } + n := 0 + if _, err := fmt.Sscanf(name[idx+2:], "%d", &n); err != nil { + return false + } + return n%2 == 0 +} + +// nil 依赖下所有便捷方法必须静默降级而非 panic。 +// +// 这是"媒体存储可关闭"在 SDK 层的对应物：内核未注入某个 API 时 +// （精简部署、插件权限不足、子进程握手尚未完成），插件的调用不该崩。 +func TestStress_NilDependenciesDegradeSilently(t *testing.T) { + s := &PluginSDK{name: "bare"} + + const workers = 16 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + s.InjectText("s", "c", "t") + s.InjectTextNoMemory("s", "c", "t") + s.InjectInterruptText("s", "c", "t") + if got := s.InjectInputSync("s", "c", "t"); got != "" { + t.Errorf("无 injector 时同步注入应返回空串，got %q", got) + } + s.InjectInputMedia("s", "c", "t", []ContentBlock{imageBlock("x")}) + if got := s.InjectInputMediaSync("s", "c", "t", nil); got != "" { + t.Errorf("无 injector 时媒体同步注入应返回空串，got %q", got) + } + s.InjectInterruptMedia("s", "c", "t", nil) + + // getter 全部应返回 nil 而非 panic + _ = s.Memory() + _ = s.TextMemory() + _ = s.DocMemory() + _ = s.Knowledge() + _ = s.LLM() + _ = s.Social() + _ = s.Events() + _ = s.PluginMgr() + _ = s.Settings() + + // 注册面无 registrar 时应返回 nil error + if err := s.RegisterTool("t", ToolDef{}, nil); err != nil { + t.Errorf("无 registrar 时 RegisterTool 应返回 nil，got %v", err) + } + if err := s.RegisterPluginAPI("a"); err != nil { + t.Errorf("无 registrar 时 RegisterPluginAPI 应返回 nil，got %v", err) + } + s.RegisterStage(StageOnInput, func(*StageContext) error { return nil }) + } + }() + } + wg.Wait() +} diff --git a/tools/plugindev/proc_runtime_test.go b/tools/plugindev/proc_runtime_test.go index 65f290f..3681612 100644 --- a/tools/plugindev/proc_runtime_test.go +++ b/tools/plugindev/proc_runtime_test.go @@ -95,12 +95,17 @@ func TestProcTemplate_CoversAllCoreMethods(t *testing.T) { // IO 注入 "io.injectText", "io.injectInterrupt", "io.injectTextNoMem", "io.injectInputSync", "io.setToolBlocks", + // 多模态注入（1.1.0 新增）。漏接线的后果是插件调 InjectInputMedia 静默无效果： + // 模板不发这个 RPC，内核也就永远收不到，而两边都不报错。 + "io.injectMedia", "io.injectMediaSync", "io.injectInterruptMedia", // 生命周期 "lifecycle.autoRestart", // 图记忆 "memory.recall", "memory.commit", "memory.introspect", "memory.merge", "memory.purge", // 文档记忆 "doc.query", "doc.insert", "doc.remove", "doc.stats", + // 文档媒体（1.1.0 新增） + "doc.insertWithMedia", // 知识库 "knowledge.search", "knowledge.add", "knowledge.list", // 文本记忆 diff --git a/tools/plugindev/templates/proc_main.go.tmpl b/tools/plugindev/templates/proc_main.go.tmpl index 537b76a..cc9fe79 100644 --- a/tools/plugindev/templates/proc_main.go.tmpl +++ b/tools/plugindev/templates/proc_main.go.tmpl @@ -490,6 +490,34 @@ func (procIO) SetToolBlocks(blocks []sdk.ContentBlock) { } } +// 带媒体的注入：插件主动发起一轮带图/音频的对话。 +// 与 SetToolBlocks 的区别是媒体在**本轮**就到模型手上，而不是等下一条 tool message。 +func (procIO) InjectInputMedia(s, c, t string, blocks []sdk.ContentBlock) { + callCoreVoid("io.injectMedia", map[string]interface{}{ + "source": s, "channel": c, "text": t, "blocks": blocks, + }) +} + +func (procIO) InjectInputMediaSync(s, c, t string, blocks []sdk.ContentBlock) string { + raw, err := callCore("io.injectMediaSync", map[string]interface{}{ + "source": s, "channel": c, "text": t, "blocks": blocks, + }) + if err != nil { + return "" + } + var r struct { + Reply string `json:"reply"` + } + json.Unmarshal(raw, &r) + return r.Reply +} + +func (procIO) InjectInterruptMedia(s, c, t string, blocks []sdk.ContentBlock) { + callCoreVoid("io.injectInterruptMedia", map[string]interface{}{ + "source": s, "channel": c, "text": t, "blocks": blocks, + }) +} + type procMemory struct{} func (procMemory) Recall(q []string, d int) ([]sdk.Entity, []sdk.Relation, error) { @@ -557,6 +585,26 @@ func (procDocMemory) Query(text string, topK int) []*sdk.Doc { func (procDocMemory) Insert(d *sdk.Doc) error { return callCoreVoid("doc.insert", map[string]interface{}{"doc": d}) } + +// InsertWithMedia 写入文档并关联媒体。 +// +// 内核会把 `[mime <短digest>] <描述>` 标记补进 Content 并挂上引用，回传的 +// doc 带着补好的 Content/ID/MediaDigests——回写进 d 让调用方能拿到这些。 +func (procDocMemory) InsertWithMedia(d *sdk.Doc, atts []sdk.MediaAttachment) error { + raw, err := callCore("doc.insertWithMedia", map[string]interface{}{ + "doc": d, "attachments": atts, + }) + if err != nil { + return err + } + var r struct { + Doc *sdk.Doc `json:"doc"` + } + if json.Unmarshal(raw, &r) == nil && r.Doc != nil { + *d = *r.Doc + } + return nil +} func (procDocMemory) Remove(id string) { callCoreVoid("doc.remove", map[string]string{"id": id}) } diff --git a/tools/plugindev/yaegi/mocksdk/plugin.go b/tools/plugindev/yaegi/mocksdk/plugin.go index 1334604..a978000 100644 --- a/tools/plugindev/yaegi/mocksdk/plugin.go +++ b/tools/plugindev/yaegi/mocksdk/plugin.go @@ -95,6 +95,29 @@ type IOInjector interface { InjectInterruptText(source, channel, text string) InjectText(source, channel, text string) InjectTextNoMemory(source, channel, text string) + // 1.1.0 媒体注入。与公共 SDK 同构：插件在 yaegi 下调得通的方法， + // 编成 plugin.bin 后必须也调得通，否则调试期与真实运行行为不一致。 + InjectInputMedia(source, channel, text string, blocks []ContentBlock) + InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string + InjectInterruptMedia(source, channel, text string, blocks []ContentBlock) + SetToolBlocks(blocks []ContentBlock) +} + +// ContentBlock 与公共 SDK 同构（OpenAI 多模态内容块格式）。 +type ContentBlock struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + ImageURL *ImageURL `json:"image_url,omitempty"` + AudioURL *AudioURL `json:"audio_url,omitempty"` +} + +type ImageURL struct { + URL string `json:"url"` + Detail string `json:"detail,omitempty"` +} + +type AudioURL struct { + URL string `json:"url"` } type EventType string @@ -159,29 +182,33 @@ type mockSettings struct{ data map[string]interface{} } func (s *mockSettings) Get(key string) (interface{}, error) { v, ok := s.data[key] - if !ok { return nil, nil } + if !ok { + return nil, nil + } return v, nil } func (s *mockSettings) Set(key string, value interface{}) error { s.data[key] = value; return nil } func (s *mockSettings) List(prefix string) ([]string, error) { var ks []string for k := range s.data { - if strings.HasPrefix(k, prefix) { ks = append(ks, k) } + if strings.HasPrefix(k, prefix) { + ks = append(ks, k) + } } return ks, nil } -func (s *mockSettings) GetCore(key string) (interface{}, error) { return nil, nil } -func (s *mockSettings) SetCore(key string, value interface{}) error { return nil } -func (s *mockSettings) ListCore(prefix string) ([]string, error) { return nil, nil } -func (s *mockSettings) GetPlugin(p, k string) (interface{}, error) { return nil, nil } -func (s *mockSettings) SetPlugin(p, k string, v interface{}) error { return nil } +func (s *mockSettings) GetCore(key string) (interface{}, error) { return nil, nil } +func (s *mockSettings) SetCore(key string, value interface{}) error { return nil } +func (s *mockSettings) ListCore(prefix string) ([]string, error) { return nil, nil } +func (s *mockSettings) GetPlugin(p, k string) (interface{}, error) { return nil, nil } +func (s *mockSettings) SetPlugin(p, k string, v interface{}) error { return nil } func (s *mockSettings) ListPlugin(p, prefix string) ([]string, error) { return nil, nil } func (s *mockSettings) RegisterDef(def ConfigDef) { logf("config def: %s = %s", def.Key, def.Default) } func (s *mockSettings) Defs(prefix string) []*ConfigDef { return nil } -func (s *mockSettings) Dump() map[string]interface{} { return s.data } -func (s *mockSettings) Plugins() []string { return nil } +func (s *mockSettings) Dump() map[string]interface{} { return s.data } +func (s *mockSettings) Plugins() []string { return nil } type Entity struct { Name string `json:"name"` @@ -195,10 +222,21 @@ type Relation struct { Object string `json:"object"` } +// Triple 与公共 SDK 同构。 +// +// ❗字段名曾是 `Predicate`，而公共 SDK 一直叫 `Relation`。 +// yaegi 解释器下插件写 `Relation:` 会报未知字段，写 `Predicate:` 则在 +// 编成 plugin.bin 时报错——谁都不对。没人发现是因为没有任何代码 +// 对着 mocksdk 编译，漂移不会被编译器抓到。 type Triple struct { - Subject string `json:"subject"` - Predicate string `json:"predicate"` - Object string `json:"object"` + Subject string `json:"subject"` + Relation string `json:"relation"` + Object string `json:"object"` + Confidence float64 `json:"confidence,omitempty"` + SubjectType string `json:"subject_type,omitempty"` + ObjectType string `json:"object_type,omitempty"` + SentenceText string `json:"sentence_text,omitempty"` + MediaDigests []string `json:"media_digests,omitempty"` } type MemoryAPI interface { @@ -212,37 +250,57 @@ type MemoryAPI interface { type mockMemory struct{} func (mockMemory) Recall(q []string, d int) ([]Entity, []Relation, error) { return nil, nil, nil } -func (mockMemory) Commit(t []Triple) error { return nil } -func (mockMemory) Introspect() (map[string]interface{}, error) { return map[string]interface{}{}, nil } -func (mockMemory) MergeEntities(s, t string) (int, error) { return 0, nil } -func (mockMemory) Purge(c map[string]string, m string) (int, error) { return 0, nil } +func (mockMemory) Commit(t []Triple) error { return nil } +func (mockMemory) Introspect() (map[string]interface{}, error) { return map[string]interface{}{}, nil } +func (mockMemory) MergeEntities(s, t string) (int, error) { return 0, nil } +func (mockMemory) Purge(c map[string]string, m string) (int, error) { return 0, nil } type Doc struct { ID string `json:"id"` Title string `json:"title"` Content string `json:"content"` Source string `json:"source"` + // 1.1.0：媒体字段。与公共 SDK 保持同构，否则插件在 yaegi 下跑得通、 + // 编成 plugin.bin 却编不过（或反之）。 + MediaDigests []string `json:"media_digests,omitempty"` + Attachments []MediaAttachment `json:"attachments,omitempty"` +} + +// MediaAttachment 与公共 SDK 同构：写入时给 Data+MIME，引用已有内容时只给 Digest。 +type MediaAttachment struct { + Digest string `json:"digest,omitempty"` + MIME string `json:"mime,omitempty"` + Data []byte `json:"data,omitempty"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` } type DocMemoryAPI interface { Query(text string, topK int) []*Doc Insert(doc *Doc) error + InsertWithMedia(doc *Doc, attachments []MediaAttachment) error Remove(id string) Stats() map[string]interface{} } type mockDocMemory struct{} -func (mockDocMemory) Query(t string, k int) []*Doc { return nil } -func (mockDocMemory) Insert(doc *Doc) error { return nil } -func (mockDocMemory) Remove(id string) {} -func (mockDocMemory) Stats() map[string]interface{} { return nil } +func (mockDocMemory) Query(t string, k int) []*Doc { return nil } +func (mockDocMemory) Insert(doc *Doc) error { return nil } +func (mockDocMemory) InsertWithMedia(doc *Doc, atts []MediaAttachment) error { + logf("doc_insert_with_media: %d 份附件", len(atts)) + return nil +} +func (mockDocMemory) Remove(id string) {} +func (mockDocMemory) Stats() map[string]interface{} { return nil } type TextEvent struct { Timestamp int64 `json:"timestamp"` Role string `json:"role"` Content string `json:"content"` Source string `json:"source"` + // 1.1.0：附件。读回时内核从正文标记反解，写入时内核把标记并进正文。 + Attachments []MediaAttachment `json:"attachments,omitempty"` } type TextMemoryAPI interface { @@ -305,8 +363,8 @@ type LLMAPI interface { type mockLLM struct{} func (mockLLM) ListSources() []string { return nil } -func (mockLLM) SetSource(n string) error { return nil } -func (mockLLM) CurrentSource() string { return "" } +func (mockLLM) SetSource(n string) error { return nil } +func (mockLLM) CurrentSource() string { return "" } type IOInjectorImpl struct{} @@ -319,27 +377,40 @@ func (IOInjectorImpl) InjectText(source, channel, text string) { func (IOInjectorImpl) InjectTextNoMemory(source, channel, text string) { logf("inject_text_no_memory: source=%s channel=%s", source, channel) } +func (IOInjectorImpl) InjectInputMedia(source, channel, text string, blocks []ContentBlock) { + logf("inject_input_media: source=%s channel=%s blocks=%d", source, channel, len(blocks)) +} +func (IOInjectorImpl) InjectInputMediaSync(source, channel, text string, blocks []ContentBlock) string { + logf("inject_input_media_sync: source=%s channel=%s blocks=%d", source, channel, len(blocks)) + return "" +} +func (IOInjectorImpl) InjectInterruptMedia(source, channel, text string, blocks []ContentBlock) { + logf("inject_interrupt_media: source=%s channel=%s blocks=%d", source, channel, len(blocks)) +} +func (IOInjectorImpl) SetToolBlocks(blocks []ContentBlock) { + logf("set_tool_blocks: blocks=%d", len(blocks)) +} type PluginSDK struct { - Name string - mu sync.RWMutex - toolDefs map[string]ToolDef - toolHandlers map[string]ToolHandler + Name string + mu sync.RWMutex + toolDefs map[string]ToolDef + toolHandlers map[string]ToolHandler stageHandlers map[string]StageHandler - outChannels map[string]ToolHandler - Settings SettingsAPI - IO IOInjector + outChannels map[string]ToolHandler + Settings SettingsAPI + IO IOInjector } func New(name string) *PluginSDK { return &PluginSDK{ - Name: name, - toolDefs: make(map[string]ToolDef), - toolHandlers: make(map[string]ToolHandler), + Name: name, + toolDefs: make(map[string]ToolDef), + toolHandlers: make(map[string]ToolHandler), stageHandlers: make(map[string]StageHandler), - outChannels: make(map[string]ToolHandler), - Settings: &mockSettings{data: map[string]interface{}{}}, - IO: IOInjectorImpl{}, + outChannels: make(map[string]ToolHandler), + Settings: &mockSettings{data: map[string]interface{}{}}, + IO: IOInjectorImpl{}, } }