feat(sdk): 多模态贯通插件边界——媒体字段、媒体注入接口与并发修复

记忆系统在核心 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 能在其上运行」的下限,媒体是可选能力。
This commit is contained in:
JianFeeeee
2026-09-06 10:03:36 +08:00
parent e256023399
commit ce5bff9275
9 changed files with 1339 additions and 97 deletions

View File

@ -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 新增)
媒体在纯文本记忆里以**标记**形式存在,格式 `[<mime> <短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 字段说明

View File

@ -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.01.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
`[<mime> <short digest>] <description>`:
```
[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

View File

@ -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 /
// InjectInterruptMediaPluginSDK 补上缺失的 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"
)

View File

@ -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.

View File

@ -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() 之前按"后注册先执行"的顺序调用,

725
sdk/stress_test.go Normal file
View File

@ -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 里是直接 fatalrecover 都接不住。
//
// 这条测试锁定的不变量:按约定持锁的并发读改写不丢更新、不 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 应返回 nilgot %v", err)
}
if err := s.RegisterPluginAPI("a"); err != nil {
t.Errorf("无 registrar 时 RegisterPluginAPI 应返回 nilgot %v", err)
}
s.RegisterStage(StageOnInput, func(*StageContext) error { return nil })
}
}()
}
wg.Wait()
}

View File

@ -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",
// 文本记忆

View File

@ -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})
}

View File

@ -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{},
}
}