7 Commits

Author SHA1 Message Date
48385858f7 fix(remotedevice): 心跳 pong 忘了 Flush —— 修「设备通道每 60 秒掉线重连」
真因(实测定位):服务端 writePong 只调 writeFrameHeader,**不 Flush**。
pong 只有两个字节,且设备空闲时没有任何别的写会顺带把 bufio 缓冲刷出去 ——
于是 pong 永远留在服务端缓冲里。

链路:客户端每 30s 发一个 ping(pingLoop)→ 服务端算出 pong 却没发出 →
客户端的读循环设的是「2 倍 ping 间隔」读超时(默认 60s)→ 每 60 秒准点
i/o timeout → 桥断开 → 3s 后重连 → 服务端 markOffline 注销 outputch,
重连后再注册。

生产日志就是这个指纹(online :20 → offline 下一分钟 :20 → 重连 :23,
连续数小时无一次例外);面板上表现为设备通道/工具凭空消失又出现,
/devices 列表跟着闪。

改法:writePong 复用 writeFrame(它 Flush)。另把客户端读循环退出时的
静默 return 改成带错误与 opcode 的日志 —— 此前断线真因在设备侧完全不可见,
只能靠对端日志倒推,正是这次排查一开始卡住的地方。

回归用例 TestWSPingGetsPongWhileIdle:只发一个 ping,随后什么都不发,
要求 2s 内必须收到 pong。**反向验证过**:把修复改回 writeFrameHeader,
用例即以 `read tcp ...: i/o timeout` 失败(与生产症状一致)。
2026-09-14 11:29:26 +08:00
bc5bfd6823 fix(scheduler): 安全点重新求值中断队列 + 抢占/背压计数修正 + 停机补终态
对照 docs/zh/input-scheduler-design.md 原文修四处(前两处是真缺陷,后两处是
观测面与设计承诺不一致),均配回归用例:

1. §4.3/§5.2「临界区结束后的第一个安全点重新求值」此前**没有实现**:
   全仓唯一的武装点是 registerInterrupt,凡被拦成「入队」的中断只能等当前任务
   自然结束。可达症状:WebUI 终止按钮连按两次,第二次落在 2s 抢占冷却窗内 →
   入队 → 再也不会被求值。修:runTaskSteps 的安全点先 rearmPending()——
   判据与 registerInterrupt 完全同一套(canPreempt + 冷却 + 临界区闸门)。

2. PreemptsByLevel 的语义是「进入 immediate 槽的次数」,但计数发生在
   setImmediateLocked 之前:immediate 是单槽,同一安全点前到达的两条同级中断里
   被降级的那条也被计成抢占。修:setImmediateLocked 只在真占住槽时返回 true,
   计数随之为真;同时把「降级入队」的责任收归调用方,消除同一任务被入队两次的
   隐患(实测该隐患会让中断任务执行两次、Executed 虚高)。

3. 状态面 Preempted 此前拿 Stats.Suspended 顶替,与 preempts_by_level 自相矛盾。
   修:Preempted = Σ PreemptsByLevel[1..4]。

4. §4.4/Q4「满时阻塞发送方 + 计数并打日志」只做了阻塞:pumpInbox 满时直接返回,
   一个字都不计。修:新增 Stats.Backpressure(+DTO 字段) 与只报一次的状态翻转日志;
   同时显式处理 enqueue 返回值(静默丢弃会让同步调用方永久挂起)。

另:Stop() 停机前排空待办——给从未运行与已挂起的、带 ResponseCh 的任务补
skipped 终态,否则 cli/clawhubadapter 这类无超时同步注入方永久挂起(§7 I5、§11.3 X4)。
emitResponse 的 ResponseCh 写入改为非阻塞 + 告警,避免一行写错就卡死调度器 goroutine。

验证:go build/vet 干净;go test -count=1 ./internal/agent/... ./internal/plugin/...
./internal/sdk/... ./cmd/... 全绿;go test -race ./internal/agent/core/ ./internal/sdk/ 干净。
新增 scheduler_rearm_test.go 六个用例(冷却期满重新求值/同级降级不计数/Preempted 求和/
停机补终态/背压计数与翻转/pumpInbox 满计数)。
2026-09-14 10:39:33 +08:00
8313120d2a chore(version): release/v1.3.x 路牌推到 1.3.12(含驻留 inputch 残留修复 + Lua events 修复) 2026-09-13 20:34:09 +08:00
5f63f6ec6c fix(lua): events.subscribe 改用内部 Subscribe + 订阅生命周期(修死锁/use-after-close)
上一版 Lua 对齐引入的 sdk.events.subscribe 有两个真问题,本提交修掉:

1) 用了公共 SDK 的 Events(),但本内核从未注入 event subscriber
   (SetEventSubscriber 全仓无调用点),拿到永远是 nil ⇒ subscribe 只会
   返回 "events unavailable"。改用内部 SDK 的 s.Subscribe——内置插件走的就是
   这条路径(cli/webui/skillmgr 全用它)。

2) 自死锁:subscribe 会在 Lua 的 plugin.start(sdk) 回调里被调用,而
   luaPlugin.Start 正持有 p.mu;原实现在 subscribe 里再 lock p.mu 追加 subs,
   不可重入 ⇒ 测试实测 30s 超时。改用独立的 subsMu。

3) use-after-close:Stop 会 Close LState,但事件订阅此前无人取消,残留回调
   再触发就会碰已关的 L。现在:Stop 先(不持 p.mu,避免与 Bus.Publish
   锁序反转)取 subsMu 取消全部订阅,再置 closed 并关 L;事件回调持 p.mu 后
   先查 closed,已进入等锁的旧回调会直接返回。

4) plugin_mgr 访问补 nil 保护(部分单测构造的 SDK 不含 pluginMgr)。

回归:TestLuaEventsSubscribeAndStopCleanup——订阅后 Publish 命中、Stop 后
再 Publish 不 panic。全套 Lua 测试在 -race 下通过。
2026-09-13 20:33:25 +08:00
a13504be38 fix(resident): 销毁驻留子时注销其入站 inputch(child/<id>)—— 修登记表脏数据累积
根因:residentInboundChannel 在 create 时把 child/<id> 登记进共享登记表
(Plugin=resident, Owner=父),但 teardownResident 只把划入的 inputch
(如 timer)归还为未分配,从未注销这条入站登记。于是每次 create/destroy
都在登记表里留下一条脏记录,且随次数单调累积。

实测(HomeAgent 侧,HΔ-Kernel v1.3.10): 后
 仍列出 child/<id>,归属 main;HomeAgent 没有任何
工具能单独注销 inputch,只能重启 homed 清掉。

修法:teardownResident 里用纯函数 inboundChannelName 算出名字并 Unregister。
不能复用 residentInboundChannel——它有重新登记的副作用。
该路径同时覆盖 destroy / reclaim / StopResidents(父退出)。

测试:TestResident_LifecycleAndNoOrphans 增加两条断言——销毁后与父退出后
child/<id> 都必须从登记表消失。
2026-09-13 20:19:47 +08:00
e75859668b chore(version): release/v1.3.x 路牌推到 1.3.11(本版内容:Lua 语义对齐 SDK 1.3.0) 2026-09-13 20:04:47 +08:00
b6a66c57fe feat(lua): Lua 插件桥全量对齐 SDK 1.3.0(媒体/注入标志位/优先级/事件/通道注销)
内核 Lua 桥(internal/plugin/lua_plugin.go)此前停在 v0.8.0 时代能力面,
1.1/1.2/1.3 新增能力只在 Go 侧存在,而 PLUGIN_DEV.md 宣称『能力完全对齐』。
本补丁把 Lua 侧补齐到与公开 SDK 1.3.0 对齐:

- 1.1 媒体:memory.commit 支持 sentence_text/media_digests;
  doc.insert_with_media + attachments;text_memory.append attachments;
  set_tool_blocks / inject_input_media(_sync) / inject_interrupt_media。
- 1.2 注入语义:inject_input_sync(_opts)、六个 *_opts 变体
  (no_memory/context_policy/cleaner_name/priority);
  ToolDef/ChannelDef 解析 context_policy。
- 1.3 优先级与动态通道:priority 常量透传;unregister_output_channel。
- StageContext 暴露 reasoning_content/context_msgs/token_usage/memory/extra/errors。
- 新增 sdk.events.subscribe 与 sdk.plugin_mgr.*。
- sdk.lua mock 同步(单一事实源在 SDK 仓 sdk/lua/sdk.lua,内核副本由
  third_party/homeagent-sdk/scripts/sync-lua-sdk.sh 同步)。

契约测试(lua_surface_test.go):
- 守住内核内嵌 mock 与 SDK 仓事实源一致;
- 守住 mock 承诺的每个函数都有运行时 RawSetString 绑定;
- 覆盖 opts/media/attachments 解析与 context_policy 透传。

文档:中英 PLUGIN_DEV.md 的 Lua API 表补齐并改为『对齐至 SDK 1.3.0』。
2026-09-13 20:04:31 +08:00
21 changed files with 1616 additions and 71 deletions

1
.gitignore vendored
View File

@ -31,6 +31,7 @@ cmd/gui/dist/
third_party/homeagent-sdk/bin/ third_party/homeagent-sdk/bin/
third_party/homeagent-sdk/tools/ third_party/homeagent-sdk/tools/
third_party/homeagent-sdk/package/ third_party/homeagent-sdk/package/
third_party/homeagent-sdk/scripts/
third_party/homeagent-sdk/.gitignore third_party/homeagent-sdk/.gitignore
third_party/homeagent-sdk/README* third_party/homeagent-sdk/README*
third_party/homeagent-sdk/example/ third_party/homeagent-sdk/example/

View File

@ -599,23 +599,26 @@ When running inside the kernel, `sdk.*` global variables are injected by the Go
### Lua SDK API ### Lua SDK API
The `sdk.*` API of Lua plugins is fully aligned with external plugins (toolchain-built `plugin.bin` subprocesses): registration functions raise a Lua error on failure; data functions uniformly return `(result, err)` with `err == nil` on success. Subsystems not wired by the core (e.g. SocialAPI) return empty values instead of errors. The `sdk.*` API of Lua plugins is aligned with external plugins (toolchain-built `plugin.bin` subprocesses) up to **SDK 1.3.0** (requires kernel **1.4.0+**, also backfilled by the Lua-alignment patch `v1.3.11`): registration functions raise a Lua error on failure; data functions uniformly return `(result, err)` with `err == nil` on success. Subsystems not wired by the core (e.g. SocialAPI) return empty values instead of errors.
> Historical note: the 1.11.3 media / inject-flags / priority capabilities were long available only on the Go side and were silently missing on the Lua side. They are now fully aligned, guarded by the contract test in `internal/plugin/lua_surface_test.go` (every function promised by the mock has a runtime binding).
**Registration** **Registration**
| Function | Description | | Function | Description |
|----------|-------------| |----------|-------------|
| `sdk.log(level, msg)` | Log output | | `sdk.log(level, msg)` | Log output |
| `sdk.register_tool(name, def, handler)` | Register tool; `def` supports `description`, `parameters`, `no_memory`, `cleaner` | | `sdk.register_tool(name, def, handler)` | Register tool; `def` supports `description`, `parameters`, `no_memory`, `context_policy` (`"none"`/`"prune"`), `cleaner` |
| `sdk.register_stage(stage, handler, scope)` | Register stage hook; `scope` is `nil`/`"global"` (default) or `"own_tools"` (fires only for `before_toolcall`/`after_toolcall` when the tool belongs to this plugin) | | `sdk.register_stage(stage, handler, scope)` | Register stage hook; `scope` is `nil`/`"global"` (default) or `"own_tools"` (fires only for `before_toolcall`/`after_toolcall` when the tool belongs to this plugin) |
| `sdk.register_api(name)` | Register API | | `sdk.register_api(name)` | Register API |
| `sdk.register_output_channel(name, caps, desc, def, handler)` | Register output channel; `def` supports `no_memory`, `cleaner` | | `sdk.register_output_channel(name, caps, desc, def, handler)` | Register output channel; `def` supports `no_memory`, `context_policy`, `cleaner` |
| `sdk.register_input_channel(name, def)` | Register input channel; `def` as above | | `sdk.register_input_channel(name, def)` | Register input channel; `def` as above |
| `sdk.unregister_output_channel(name)` | Unregister an output channel (for resource-bound channels, e.g. remote devices); returns `(nil, err)` |
| `sdk.set_auto_restart(enabled)` | Auto-restart the plugin after a crash | | `sdk.set_auto_restart(enabled)` | Auto-restart the plugin after a crash |
**Stage hook context** **Stage hook context**
Stage handlers receive the full context (same as external plugins): `raw_message`, `user_id`, `group_id`, `phase`, `llm_text`, `final_text`, `no_memory`, `response` (when responded), `tool_calls`, `tool_results`. Stage handlers receive the full context (same as external plugins): `raw_message`, `user_id`, `group_id`, `phase`, `llm_text`, `reasoning_content`, `final_text`, `no_memory`, `context_msgs`, `token_usage`, `memory`, `extra`, `errors`, `response` (when responded), `tool_calls`, `tool_results`.
**Stage writeback**: the `ctx` table passed to the handler is a reference — mutating writable fields inside the handler syncs back to the core `StageContext` (aligned with subprocess external-plugin capability): **Stage writeback**: the `ctx` table passed to the handler is a reference — mutating writable fields inside the handler syncs back to the core `StageContext` (aligned with subprocess external-plugin capability):
@ -644,20 +647,37 @@ Writable fields: `raw_message`, `llm_text`, `final_text`, `user_id`, `group_id`,
| `sdk.inject_text(source, channel, text)` | Deliver text message | | `sdk.inject_text(source, channel, text)` | Deliver text message |
| `sdk.inject_interrupt(source, channel, text)` | Interrupt delivery | | `sdk.inject_interrupt(source, channel, text)` | Interrupt delivery |
| `sdk.inject_text_no_memory(source, channel, text)` | Deliver without memory computation | | `sdk.inject_text_no_memory(source, channel, text)` | Deliver without memory computation |
| `sdk.inject_text_opts` / `sdk.inject_interrupt_opts(source, channel, text, opts)` | Delivery with flags; `opts = { no_memory=bool, context_policy="none"|"prune", cleaner_name=string, priority="L1".."L3" }` |
| `sdk.inject_input_sync(source, channel, text)` | Inject synchronously and wait for this turn's reply; returns `(reply, err)`, reply is nil when there is none |
| `sdk.inject_input_sync_opts(source, channel, text, opts)` | Same, with flags |
| `sdk.inject_input_media(source, channel, text, blocks)` | Inject text + multimodal content blocks |
| `sdk.inject_input_media_opts(source, channel, text, blocks, opts)` | Same, with flags |
| `sdk.inject_input_media_sync` / `..._sync_opts(...)` | Synchronous media injection; returns `(reply, err)` |
| `sdk.inject_interrupt_media(source, channel, text, blocks)` | Interrupt delivery with media |
| `sdk.inject_interrupt_media_opts(source, channel, text, blocks, opts)` | Same, with flags |
| `sdk.set_tool_blocks(blocks)` | Set multimodal blocks carried by the next tool message (lets the model see images / hear audio) |
Each `blocks` item: `{ type="text", text="..." }`, `{ type="image_url", image_url={ url="...", detail="high" } }`, or `{ type="audio_url", audio_url={ url="..." } }`. An absent `opts` is the zero value (recorded in memory + no pruning), equivalent to the three-argument form.
**Data APIs (aligned with subprocess external plugins, all return `(result, err)`)** **Data APIs (aligned with subprocess external plugins, all return `(result, err)`)**
| Sub-table | Functions | | Sub-table | Functions |
|-----------|-----------| |-----------|-----------|
| `sdk.memory.*` | `recall(query, depth)`, `commit({triples})`, `introspect()`, `merge(source, target)`, `purge(criteria, hard)` | | `sdk.memory.*` | `recall(query, depth)`, `commit({triples})` (triple supports `subject/relation/object/confidence/subject_type/object_type/sentence_text/media_digests`), `introspect()`, `merge(source, target)`, `purge(criteria, hard)` |
| `sdk.doc.*` | `query(text, top_k)`, `insert({id,title,content})`, `remove(id)`, `stats()` | | `sdk.doc.*` | `query(text, top_k)`, `insert({id,title,content})`, `insert_with_media(doc, attachments)`, `remove(id)`, `stats()` |
| `sdk.knowledge.*` | `search(query, limit)`, `add(tag, content)`, `list()` | | `sdk.knowledge.*` | `search(query, limit)`, `add(tag, content)`, `list()` |
| `sdk.text_memory.*` | `append({role,content,timestamp,channel})` | | `sdk.text_memory.*` | `append({role,content,timestamp,channel,attachments})` |
| `sdk.llm.*` | `list_sources()`, `set_source(name)`, `current_source()` | | `sdk.llm.*` | `list_sources()`, `set_source(name)`, `current_source()` |
| `sdk.social.*` (read-only) | `get_person(name)`, `get_network(name, depth)`, `get_trait(name, trait)`, `get_relations(name)`, `list_persons()` | | `sdk.social.*` (read-only) | `get_person(name)`, `get_network(name, depth)`, `get_trait(name, trait)`, `get_relations(name)`, `list_persons()` |
| `sdk.events.*` | `subscribe(event_type, handler)` → returns an unsubscribe function; handler receives `{type,source,timestamp,payload}` |
| `sdk.plugin_mgr.*` | `reload_one(name)`, `list_loaded()`, `is_disabled(name)` |
| `sdk.json.*` | `encode(val)`, `decode(str)` | | `sdk.json.*` | `encode(val)`, `decode(str)` |
| `sdk.http.*` | `get(url)`, `post(url, body, content_type)` | | `sdk.http.*` | `get(url)`, `post(url, body, content_type)` |
Each `attachments` item: `{ digest=, mime=, name=, data=<base64> }`; with `data` it is new content (stored in the content-addressed store), with only `digest` it references existing content.
> The `sdk.events.subscribe` callback runs on the kernel's event-publishing goroutine, and Lua is single-state + mutex-guarded — **do only lightweight forwarding inside the callback; never block**, or every call of this plugin will stall.
--- ---
<img src="../../assets/branding/mascot-xiaozhai.webp" width="20" style="border-radius:50%;vertical-align:middle"> : <img src="../../assets/branding/mascot-xiaozhai.webp" width="20" style="border-radius:50%;vertical-align:middle"> :

View File

@ -592,23 +592,26 @@ lua main.lua
### Lua SDK API ### Lua SDK API
Lua 插件的 `sdk.*` API 与外部插件(工具链编译的 `plugin.bin` 子进程)能力完全对齐:注册类函数调用即时报错(抛 Lua error数据类函数统一返回 `(result, err)``err` 为 nil 表示成功。核心未装配的子系统(如 SocialAPI返回空值而非报错。 Lua 插件的 `sdk.*` API 与外部插件(工具链编译的 `plugin.bin` 子进程)能力对齐**SDK 1.3.0**(需内核 **1.4.0+**,也在 `v1.3.11` 的 Lua 对齐补丁中回填):注册类函数调用即时报错(抛 Lua error数据类函数统一返回 `(result, err)``err` 为 nil 表示成功。核心未装配的子系统(如 SocialAPI返回空值而非报错。
> 历史提醒1.11.3 的媒体/注入标志位/优先级能力曾长期只在 Go 侧Lua 侧静默缺失。现已全量对齐,并由 `internal/plugin/lua_surface_test.go` 的契约测试守住「mock 承诺的每个函数都有运行时绑定」。
**注册类** **注册类**
| 函数 | 说明 | | 函数 | 说明 |
|------|------| |------|------|
| `sdk.log(level, msg)` | 日志输出 | | `sdk.log(level, msg)` | 日志输出 |
| `sdk.register_tool(name, def, handler)` | 注册工具;`def` 支持 `description``parameters``no_memory``cleaner` | | `sdk.register_tool(name, def, handler)` | 注册工具;`def` 支持 `description``parameters``no_memory``context_policy``"none"`/`"prune"`)、`cleaner` |
| `sdk.register_stage(stage, handler, scope)` | 注册阶段钩子;`scope``nil`/`"global"`(默认)或 `"own_tools"`(仅 `before_toolcall`/`after_toolcall` 且工具属于本插件时触发) | | `sdk.register_stage(stage, handler, scope)` | 注册阶段钩子;`scope``nil`/`"global"`(默认)或 `"own_tools"`(仅 `before_toolcall`/`after_toolcall` 且工具属于本插件时触发) |
| `sdk.register_api(name)` | 注册 API | | `sdk.register_api(name)` | 注册 API |
| `sdk.register_output_channel(name, caps, desc, def, handler)` | 注册输出通道;`def` 支持 `no_memory``cleaner` | | `sdk.register_output_channel(name, caps, desc, def, handler)` | 注册输出通道;`def` 支持 `no_memory``context_policy``cleaner` |
| `sdk.register_input_channel(name, def)` | 注册输入通道;`def` 同上 | | `sdk.register_input_channel(name, def)` | 注册输入通道;`def` 同上 |
| `sdk.unregister_output_channel(name)` | 注销输出通道(随资源生灭的动态通道,如远程设备);返回 `(nil, err)` |
| `sdk.set_auto_restart(enabled)` | 崩溃时内核自动拉起插件 | | `sdk.set_auto_restart(enabled)` | 崩溃时内核自动拉起插件 |
**阶段钩子上下文** **阶段钩子上下文**
`register_stage` 的 handler 收到完整上下文(与外部插件一致):`raw_message``user_id``group_id``phase``llm_text``final_text``no_memory``response`(已响应时)、`tool_calls``tool_results` `register_stage` 的 handler 收到完整上下文(与外部插件一致):`raw_message``user_id``group_id``phase``llm_text``reasoning_content``final_text``no_memory``context_msgs``token_usage``memory``extra``errors``response`(已响应时)、`tool_calls``tool_results`
**Stage 写回**handler 收到的 `ctx` 是引用 table——在 handler 内直接修改可写回字段并同步至内核 `StageContext`(与子进程外部插件能力对齐): **Stage 写回**handler 收到的 `ctx` 是引用 table——在 handler 内直接修改可写回字段并同步至内核 `StageContext`(与子进程外部插件能力对齐):
@ -637,20 +640,37 @@ end)
| `sdk.inject_text(source, channel, text)` | 投递文本消息 | | `sdk.inject_text(source, channel, text)` | 投递文本消息 |
| `sdk.inject_interrupt(source, channel, text)` | 中断投递 | | `sdk.inject_interrupt(source, channel, text)` | 中断投递 |
| `sdk.inject_text_no_memory(source, channel, text)` | 免记忆投递 | | `sdk.inject_text_no_memory(source, channel, text)` | 免记忆投递 |
| `sdk.inject_text_opts` / `sdk.inject_interrupt_opts(source, channel, text, opts)` | 带标志位投递;`opts = { no_memory=bool, context_policy="none"|"prune", cleaner_name=string, priority="L1".."L3" }` |
| `sdk.inject_input_sync(source, channel, text)` | 同步注入并等本轮回复;返回 `(reply, err)`,无回复时 reply 为 nil |
| `sdk.inject_input_sync_opts(source, channel, text, opts)` | 同上带标志位 |
| `sdk.inject_input_media(source, channel, text, blocks)` | 注入文本 + 多模态内容块 |
| `sdk.inject_input_media_opts(source, channel, text, blocks, opts)` | 同上带标志位 |
| `sdk.inject_input_media_sync` / `..._sync_opts(...)` | 带媒体的同步注入;返回 `(reply, err)` |
| `sdk.inject_interrupt_media(source, channel, text, blocks)` | 带媒体的中断注入 |
| `sdk.inject_interrupt_media_opts(source, channel, text, blocks, opts)` | 同上带标志位 |
| `sdk.set_tool_blocks(blocks)` | 设置下一轮 tool message 携带的多模态内容块(模型据此看图/听音频) |
`blocks` 每项形如:`{ type="text", text="..." }``{ type="image_url", image_url={ url="...", detail="high" } }``{ type="audio_url", audio_url={ url="..." } }``opts` 缺省即零值(记入记忆 + 不裁剪),与三参数版本等价。
**数据类(与子进程外部插件对齐,均返回 `(result, err)`** **数据类(与子进程外部插件对齐,均返回 `(result, err)`**
| 子表 | 函数 | | 子表 | 函数 |
|------|------| |------|------|
| `sdk.memory.*` | `recall(query, depth)``commit({triples})``introspect()``merge(source, target)``purge(criteria, hard)` | | `sdk.memory.*` | `recall(query, depth)``commit({triples})`triple 支持 `subject/relation/object/confidence/subject_type/object_type/sentence_text/media_digests``introspect()``merge(source, target)``purge(criteria, hard)` |
| `sdk.doc.*` | `query(text, top_k)``insert({id,title,content})``remove(id)``stats()` | | `sdk.doc.*` | `query(text, top_k)``insert({id,title,content})``insert_with_media(doc, attachments)``remove(id)``stats()` |
| `sdk.knowledge.*` | `search(query, limit)``add(tag, content)``list()` | | `sdk.knowledge.*` | `search(query, limit)``add(tag, content)``list()` |
| `sdk.text_memory.*` | `append({role,content,timestamp,channel})` | | `sdk.text_memory.*` | `append({role,content,timestamp,channel,attachments})` |
| `sdk.llm.*` | `list_sources()``set_source(name)``current_source()` | | `sdk.llm.*` | `list_sources()``set_source(name)``current_source()` |
| `sdk.social.*`(只读) | `get_person(name)``get_network(name, depth)``get_trait(name, trait)``get_relations(name)``list_persons()` | | `sdk.social.*`(只读) | `get_person(name)``get_network(name, depth)``get_trait(name, trait)``get_relations(name)``list_persons()` |
| `sdk.events.*` | `subscribe(event_type, handler)` → 返回取消订阅函数handler 收到 `{type,source,timestamp,payload}` |
| `sdk.plugin_mgr.*` | `reload_one(name)``list_loaded()``is_disabled(name)` |
| `sdk.json.*` | `encode(val)``decode(str)` | | `sdk.json.*` | `encode(val)``decode(str)` |
| `sdk.http.*` | `get(url)``post(url, body, content_type)` | | `sdk.http.*` | `get(url)``post(url, body, content_type)` |
`attachments` 每项:`{ digest=, mime=, name=, data=<base64> }`;带 `data` 是新内容(落进内容寻址存储),只带 `digest` 是引用已有内容。
> `sdk.events.subscribe` 的回调在内核事件发布 goroutine 上执行,且 Lua 是单状态 + 互斥锁——**回调内只做轻量转发,不可阻塞**,否则会卡死本插件的全部调用。
--- ---
<img src="../../assets/branding/mascot-xiaozhai.webp" width="20" style="border-radius:50%;vertical-align:middle"> : <img src="../../assets/branding/mascot-xiaozhai.webp" width="20" style="border-radius:50%;vertical-align:middle"> :

View File

@ -404,9 +404,30 @@ func (a *Agent) Start() {
func (a *Agent) Stop() { func (a *Agent) Stop() {
// 父退出**必须**销毁全部驻留子(设计 §10 硬约束:子不得比父活得久、不留孤儿)。 // 父退出**必须**销毁全部驻留子(设计 §10 硬约束:子不得比父活得久、不留孤儿)。
a.StopResidents() a.StopResidents()
// 停机前给待办任务补终态。运行中的任务会经 cancel → LLM 失败 → emitResponse
// 自然拿到终态,但**从未运行**(排队/待处理)与**已挂起**的任务不会有任何人
// 回它们;带 ResponseCh 的同步注入方cli / clawhubadapter 均无超时)会永久挂起
// (设计 §7 I5、§11.3 X2/X4。必须在 cancel 之前做cancel 会让调度器直接 return。
a.drainPendingInterrupts("agent_stopped")
a.cancel() a.cancel()
} }
// drainPendingInterrupts 给排队/待处理/已挂起任务中带同步回执通道的调用方补一条
// skipped 终态(复用 emitSkippedReply非阻塞写不对外发 agent_output 事件)。
func (a *Agent) drainPendingInterrupts(reason string) {
if a.sched == nil {
return
}
pending := a.sched.pendingEvents()
if len(pending) == 0 {
return
}
for _, evt := range pending {
a.emitSkippedReply(evt, reason)
}
log.Printf("[agent] %s: 停机,%d 条待办任务已补 skipped 终态", a.id, len(pending))
}
// graphMemoryOf 决定本 agent 的图记忆共同面实现。 // graphMemoryOf 决定本 agent 的图记忆共同面实现。
// //
// - 轻量内核(给了 LightMemory用 LightMemory**整理面保持 nil** // - 轻量内核(给了 LightMemory用 LightMemory**整理面保持 nil**

View File

@ -331,13 +331,20 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
payload["usage"] = stageCtx.TokenUsage payload["usage"] = stageCtx.TokenUsage
} }
if evt.ResponseCh != nil { if evt.ResponseCh != nil {
evt.ResponseCh <- &agentIO.OutputEvent{ // 非阻塞写:ResponseCh 由同步调用方以 cap=1 创建。按不变量 I5每任务恰一次
// 终态)这里永远写得进去;但一旦哪天写出第二次,阻塞会卡死**调度器 goroutine**
// (整个 agent 停摆),而丢弃只是丢一条回执——与 emitSkippedReply 对称。
select {
case evt.ResponseCh <- &agentIO.OutputEvent{
RequestID: evt.RequestID, RequestID: evt.RequestID,
Target: evt.Source, Target: evt.Source,
Type: "text", Type: "text",
Payload: payload, Payload: payload,
Done: true, Done: true,
OutputChannel: ch, OutputChannel: ch,
}:
default:
log.Printf("[agent] ResponseCh 已满终态回执被丢弃request=%s可能违反不变量 I5", evt.RequestID)
} }
} }

View File

@ -198,8 +198,13 @@ func (a *Agent) SpawnResident(opts ResidentOptions) (ResidentInfo, error) {
func (a *Agent) residentParentSource() string { return "parent/" + string(a.id) } func (a *Agent) residentParentSource() string { return "parent/" + string(a.id) }
// residentInboundChannel 是"父接收某个子的消息"的 inputch 名(登记进登记表可见)。 // residentInboundChannel 是"父接收某个子的消息"的 inputch 名(登记进登记表可见)。
//
// inboundChannelName 只拼名字,不产生副作用(注销路径要用它算出同一个名字,
// 不能再去调 residentInboundChannel——那会顺手把刚摘掉的登记又写回去
func inboundChannelName(childID string) string { return "child/" + childID }
func (a *Agent) residentInboundChannel(childID string) string { func (a *Agent) residentInboundChannel(childID string) string {
ch := "child/" + childID ch := inboundChannelName(childID)
if reg := a.io.ChannelRegistry(); reg != nil { if reg := a.io.ChannelRegistry(); reg != nil {
// 归属父自己:它是父的入站 inputch。 // 归属父自己:它是父的入站 inputch。
_ = reg.Register(agentIO.InputChannel{Name: ch, Plugin: "resident", Owner: string(a.id)}) _ = reg.Register(agentIO.InputChannel{Name: ch, Plugin: "resident", Owner: string(a.id)})
@ -246,6 +251,14 @@ func (a *Agent) teardownResident(rc *residentChild) {
for _, ch := range rc.inputChs { for _, ch := range rc.inputChs {
_ = reg.Assign(ch, "", 0) _ = reg.Assign(ch, "", 0)
} }
// 注销"父接收该子消息"的入站 inputchchild/<id>)。
//
// 它由 residentInboundChannel 在 create 时登记Owner=父),销毁时必须
// 一并摘掉:登记表是共享的、按 name 全局唯一,残留会随 create/destroy
// 次数单调累积脏数据。实测destroy 后 child/<id> 仍挂在根 agent 名下,
// 而外部没有任何工具能单独注销 inputch只能重启 homed 清。
// 注意用纯函数算名字,不要再走 residentInboundChannel会重新登记
reg.Unregister(inboundChannelName(rc.id))
} }
} }

View File

@ -142,6 +142,11 @@ func TestResident_LifecycleAndNoOrphans(t *testing.T) {
if ch, _ := reg.Lookup("sub/in"); ch.Owner != "" { if ch, _ := reg.Lookup("sub/in"); ch.Owner != "" {
t.Fatalf("销毁后 inputch 应回到未分配:%+v", ch) t.Fatalf("销毁后 inputch 应回到未分配:%+v", ch)
} }
// 父的入站 inputchchild/<id>)必须在销毁时一并注销,否则登记表残留脏数据——
// HomeAgent 实测destroy 后 child/<id> 仍挂在根 agent 名下,且无工具可单独注销。
if inbound, ok := reg.Lookup("child/child-1"); ok {
t.Fatalf("销毁后父的入站 inputch 应被注销,实际残留:%+v", inbound)
}
if err := parent.DestroyResident("child-1"); err == nil { if err := parent.DestroyResident("child-1"); err == nil {
t.Fatal("重复销毁应报错") t.Fatal("重复销毁应报错")
} }
@ -159,6 +164,9 @@ func TestResident_LifecycleAndNoOrphans(t *testing.T) {
if _, err := osStat(filepath.Join(dir, "residents", id)); err == nil { if _, err := osStat(filepath.Join(dir, "residents", id)); err == nil {
t.Fatalf("子 %s 的 temp 目录应被丢弃", id) t.Fatalf("子 %s 的 temp 目录应被丢弃", id)
} }
if _, ok := reg.Lookup(inboundChannelName(id)); ok {
t.Fatalf("父退出后子 %s 的入站 inputch 应被注销", id)
}
} }
} }

View File

@ -229,6 +229,10 @@ type SchedulerStats struct {
Executed uint64 Executed uint64
// Rejected 是因队列满(或深度超限)而未被接纳的次数。 // Rejected 是因队列满(或深度超限)而未被接纳的次数。
Rejected uint64 Rejected uint64
// Backpressure 是就绪队列满、输入被挡回 channel 的次数
// (设计 §4.4 / §11.4 Q4满时阻塞发送方**必须计数并打日志**)。
// 与 Rejected 的区别Rejected 是「丢了」Backpressure 是「暂时不收、发送方在等」。
Backpressure uint64
// Suspended / Resumed 是挂起与恢复的次数。 // Suspended / Resumed 是挂起与恢复的次数。
// 不变量:系统排空后 Suspended == Resumed挂起必然被恢复 // 不变量:系统排空后 Suspended == Resumed挂起必然被恢复
// 因此两者各自只在**一处**计数suspend / resumeTask // 因此两者各自只在**一处**计数suspend / resumeTask
@ -284,7 +288,13 @@ func (a *Agent) schedulerStatus() sdk.SchedulerStatus {
Rejected: snap.Stats.Rejected, Rejected: snap.Stats.Rejected,
Suspended: snap.Stats.Suspended, Suspended: snap.Stats.Suspended,
Resumed: snap.Stats.Resumed, Resumed: snap.Stats.Resumed,
Preempted: snap.Stats.Suspended, Backpressure: snap.Stats.Backpressure,
}
// Preempted 是「各级抢占成功次数之和」,**不是** Suspended受害者可能在
// 让位信号生效前就自行结束,此时有抢占而没有挂起(见 PreemptsByLevel 注释)。
// 此前这里直接拿 Suspended 顶替,导致 DTO 里 preempted 与 preempts_by_level 自相矛盾。
for lv := LevelBackground; lv <= LevelCritical; lv++ {
out.Preempted += snap.Stats.PreemptsByLevel[lv]
} }
if snap.Running != nil { if snap.Running != nil {
out.Running = &sdk.SchedulerTask{ out.Running = &sdk.SchedulerTask{
@ -320,6 +330,9 @@ type scheduler struct {
// critical 报告运行任务是否在不可抢占临界区(如记忆整理)。 // critical 报告运行任务是否在不可抢占临界区(如记忆整理)。
// 由于 interceptLoop 要读它,必须是原子的:帧仍只由调度器读写。 // 由于 interceptLoop 要读它,必须是原子的:帧仍只由调度器读写。
critical atomic.Bool critical atomic.Bool
// backpressured 记录「就绪队列满」这一状态的翻转,用于只打一次日志。
// 满着的时候 pumpInbox 每轮都会走到,逐轮打日志会把日志刷爆。
backpressured bool
// wake 用于把空闲的调度器叫醒pendingInterrupts 不是 channel // wake 用于把空闲的调度器叫醒pendingInterrupts 不是 channel
// 没有这个信号时“空闲时到达的中断”会一直等下一次输入(设计 §5.1 ③)。 // 没有这个信号时“空闲时到达的中断”会一直等下一次输入(设计 §5.1 ③)。
wake chan struct{} wake chan struct{}
@ -381,6 +394,28 @@ func (s *scheduler) hasRoom() bool {
return len(s.queue) < s.maxQueue return len(s.queue) < s.maxQueue
} }
// noteBackpressure 记一次背压,并报告这是否是「从有空间 → 满」的翻转。
//
// 为什么需要翻转信息:满的时候每轮泵入都会调用本函数,逐轮打日志会刷爆;
// 而设计 §4.4 要求「必须计数并打日志」——两者靠这个布尔量同时满足。
func (s *scheduler) noteBackpressure() bool {
s.mu.Lock()
defer s.mu.Unlock()
s.stats.Backpressure++
if s.backpressured {
return false
}
s.backpressured = true
return true
}
// clearBackpressure 在就绪队列重新可收(泵空)时复位翻转标记。
func (s *scheduler) clearBackpressure() {
s.mu.Lock()
defer s.mu.Unlock()
s.backpressured = false
}
// allocateIDLocked 分配任务 ID 与入队时刻(调用方持锁)。 // allocateIDLocked 分配任务 ID 与入队时刻(调用方持锁)。
func (s *scheduler) allocateIDLocked(t *Task) { func (s *scheduler) allocateIDLocked(t *Task) {
s.seq++ s.seq++
@ -490,10 +525,22 @@ func (s *scheduler) interruptCountLocked() int {
// setImmediateLocked 登记一个应“立即运行”的抢占者。 // setImmediateLocked 登记一个应“立即运行”的抢占者。
// //
// 槽只有一格:若已有抢占者且新的级别更高,旧的降级入队;否则新的入队。 // 槽只有一格:若已有抢占者且新的级别更高,旧的降级入队;否则新的入队。
func (s *scheduler) setImmediateLocked(t *Task) { // 返回 true 表示 t **确实占住了 immediate 槽**false 表示它被降级进了自己的
// 级别队列immediate 是单槽,这是设计要求的降级分支,见设计 §2「至多一个」
//
// 调用方必须用返回值决定是否计入 PreemptsByLevel那条计数器的语义是
// 「进入 immediate 的次数」,被降级的中断从未进过 immediate。
// setImmediateLocked 尝试把 t 放进 immediate 槽。
//
// 返回 truet 已占住 immediate若原有抢占者被顶掉它**已被**降级入队)。
// 返回 falset 没有进 immediate且本函数**未动 t** —— 调用方负责按级别入队。
//
// 把「降级入队」的责任留给调用方,是为了让「到底入队了几次」只有一个出口:
// 早先由本函数在返回 false 前自行入队,调用方又照着 false 再入一次,
// 同一任务就会在队列里出现两份实测中断任务被执行两次、Executed 虚高)。
func (s *scheduler) setImmediateLocked(t *Task) bool {
if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) { if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) {
s.enqueueInterruptLocked(t) return false
return
} }
if s.immediate != nil { if s.immediate != nil {
s.enqueueInterruptLocked(s.immediate) s.enqueueInterruptLocked(s.immediate)
@ -501,6 +548,7 @@ func (s *scheduler) setImmediateLocked(t *Task) {
s.allocateIDLocked(t) s.allocateIDLocked(t)
s.stats.Enqueued++ s.stats.Enqueued++
s.immediate = t s.immediate = t
return true
} }
func removeTask(list []*Task, target *Task) []*Task { func removeTask(list []*Task, target *Task) []*Task {
@ -571,11 +619,16 @@ func (s *scheduler) registerInterrupt(t *Task) bool {
arm := false arm := false
if !critical && canPreempt(t, running) { if !critical && canPreempt(t, running) {
if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown {
arm = true // 只有**真的占住 immediate 槽**才算一次抢占,才计入 PreemptsByLevel
s.preemptArmed = true // immediate 是单槽若它被另一个更高级的抢占者占着t 会走上而下的
s.preemptLevel = t.Level // 「否则入队」分支——那种情况 t 从未进入 immediate否则同一安全点前
s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level) // 到达两条同级中断时该计数会高估)。
s.setImmediateLocked(t) if s.setImmediateLocked(t) {
arm = true
s.preemptArmed = true
s.preemptLevel = t.Level
s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
}
} }
} }
if !arm { if !arm {
@ -606,6 +659,54 @@ func (s *scheduler) clearPreempt() {
s.mu.Unlock() s.mu.Unlock()
} }
// rearmPending 在**安全点重新求值**中断队列(设计 §4.3 / §5.2)。
//
// 为什么必须有这一步:中断只在 registerInterrupt 里被武装一次,而那一刻运行任务
// 可能正在临界区S_TOOL_EXEC / ONNX / CAS或处于抢占冷却期于是请求只能入队。
// 若安全点不再回头看队列,它就永远等不到执行——只能等当前任务**自然结束**
// 这违背设计承诺的「临界区期间到达的抢占请求……在临界区结束后的第一个安全点
// 重新求值」。可复现症状WebUI 终止按钮连按两次,第二次(落在 2s 冷却窗内)
// 入队后再也不会被求值,「终止」看起来没反应。
//
// 判据与 registerInterrupt **完全同一套**canPreempt + 冷却 + 临界区闸门),
// 因此不会凭空制造设计之外的抢占。
func (s *scheduler) rearmPending() {
s.mu.Lock()
defer s.mu.Unlock()
// 已有让位信号、或 immediate 槽已被占用:下一个安全点的选择已经在路上,
// 不必(也不该)重复武装。
if s.preemptArmed || s.immediate != nil || s.running == nil || s.critical.Load() {
return
}
// 冷却期内不武装:与 registerInterrupt 同一判据(抗饥饿)。
if !s.running.LastPreemptAt.IsZero() && time.Since(s.running.LastPreemptAt) < preemptCooldown {
return
}
// 中断队列本就按级别组织:从最高级往下找第一条能抢占的队头。
// (队列里的任务有效级恒等于基础级,故「第一条能抢」= 最高级可抢占者。)
for lv := LevelCritical; lv >= LevelBackground; lv-- {
q := s.interruptQueues[lv]
if len(q) == 0 {
continue
}
t := q[0]
if !canPreempt(t, s.running) {
continue
}
s.popInterruptLocked(lv)
if s.setImmediateLocked(t) {
s.preemptArmed = true
s.preemptLevel = t.Level
s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
} else {
// immediate 槽没拿到(理论上进不来,顶部已判 immediate == nil放回队列
// 否则任务会凭空消失。
s.enqueueInterruptLocked(t)
}
return
}
}
// suspend 保存现场。 // suspend 保存现场。
// //
// 深度上界是**结构推论**= 中断级数),不是配置项:安全点上的 canSuspend 已提前 // 深度上界是**结构推论**= 中断级数),不是配置项:安全点上的 canSuspend 已提前
@ -638,6 +739,37 @@ func (s *scheduler) canSuspend() bool {
return len(s.suspendStack) < s.maxInterruptFrames return len(s.suspendStack) < s.maxInterruptFrames
} }
// pendingEvents 收集**尚未执行**(排队队列 / 四条中断队列 / immediate
// **已挂起**(中断栈)任务所携带的、且带同步回执通道的输入事件。
//
// 用途只有一个:停机收尾。这些任务不会再被调度,若不给它们补终态,
// 无超时的同步注入方cli / clawhubadapter会永久挂起设计 §7 I5、§11.3 X2/X4
func (s *scheduler) pendingEvents() []*agentIO.InputEvent {
s.mu.Lock()
defer s.mu.Unlock()
var out []*agentIO.InputEvent
add := func(t *Task) {
if t != nil && t.Event != nil && t.Event.ResponseCh != nil {
out = append(out, t.Event)
}
}
for _, t := range s.queue {
add(t)
}
for lv := LevelBackground; lv <= LevelCritical; lv++ {
for _, t := range s.interruptQueues[lv] {
add(t)
}
}
add(s.immediate)
for _, f := range s.suspendStack {
if f != nil {
add(f.Task)
}
}
return out
}
// done 标记任务执行结束。 // done 标记任务执行结束。
func (s *scheduler) done(t *Task) { func (s *scheduler) done(t *Task) {
s.mu.Lock() s.mu.Lock()
@ -796,9 +928,11 @@ func (a *Agent) schedulerLoop() {
// 无待办阻塞等新输入、新中断wake或退出。 // 无待办阻塞等新输入、新中断wake或退出。
select { select {
case evt := <-a.io.InputChan(): case evt := <-a.io.InputChan():
a.sched.enqueue(newInputTask(evt)) if !a.sched.enqueue(newInputTask(evt)) {
a.emitSkippedReply(evt, "queue_full")
}
case msg := <-a.selfInputCh: case msg := <-a.selfInputCh:
a.sched.enqueue(newSelfTask(msg)) _ = a.sched.enqueue(newSelfTask(msg))
case <-a.sched.wake: case <-a.sched.wake:
// 中断已入 pendingInterrupts回到循环顶部重新挑选。 // 中断已入 pendingInterrupts回到循环顶部重新挑选。
case <-a.ctx.Done(): case <-a.ctx.Done():
@ -825,15 +959,28 @@ func (a *Agent) pumpInbox() {
for a.sched.hasRoom() { for a.sched.hasRoom() {
select { select {
case evt := <-a.io.InputChan(): case evt := <-a.io.InputChan():
a.sched.enqueue(newInputTask(evt)) // 返回值必须处理:静默丢弃会让同步调用方永久挂起(回执路径 E
if !a.sched.enqueue(newInputTask(evt)) {
a.sched.noteBackpressure()
a.emitSkippedReply(evt, "queue_full")
}
case msg := <-a.selfInputCh: case msg := <-a.selfInputCh:
a.sched.enqueue(newSelfTask(msg)) // 自循环输入没有同步调用方,满时记一次背压即可。
if !a.sched.enqueue(newSelfTask(msg)) {
a.sched.noteBackpressure()
}
case <-a.ctx.Done(): case <-a.ctx.Done():
return return
default: default:
a.sched.clearBackpressure()
return return
} }
} }
// 队列满:输入留在 channel 里,发送方阻塞(设计 §4.4「阻塞发送方」)。
// 必须计数并打日志——否则运维看到 Rejected=0 会以为没背压,而输入正卡在 channel。
if a.sched.noteBackpressure() {
log.Printf("[agent] ready queue full (%d), input channel backpressured", a.sched.maxQueue)
}
} }
// executeTask 执行一个任务(测试与旧调用方的入口);见 executeNewTask。 // executeTask 执行一个任务(测试与旧调用方的入口);见 executeNewTask。

View File

@ -0,0 +1,186 @@
package core
// 回归测试:安全点「重新求值」、抢占计数语义、停机补终态、背压计数。
//
// 对照设计稿原文修正的四条:
//
// 1. §4.3/§5.2 —— 临界区期间到达的抢占请求「不丢失:按级别进入中断队列,
// 在**临界区结束后的第一个安全点重新求值**」。实现里此前没有这一步:
// 唯一的武装点是 registerInterrupt凡被拦成「入队」的中断只能等当前任务
// **自然结束**。可复现症状WebUI 终止按钮连按两次,第二次落在 2s 抢占冷却
// 窗内 → 入队 → 再也不会被求值,「终止」看起来没反应。
// 2. SchedulerStats.PreemptsByLevel 的语义是「判定可抢占**并进入 immediate** 的
// 次数」;此前在 setImmediateLocked 之前就计数,于是同一安全点前到达的两条同级
// 中断里、被降级入队的那条也被计入immediate 是单槽,降级是设计要求的路径)。
// 3. 状态面 Preempted 此前直接拿 Stats.Suspended 顶替,与 preempts_by_level 自相矛盾。
// 4. §4.4 / §11.4 Q4 —— 就绪队列满必须「阻塞发送方 + **计数并打日志**」。
import (
"testing"
"time"
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
)
// 被冷却拦成入队的中断,在冷却期满后的第一个安全点必须被重新武装。
//
// 这是「终止按钮连按两次」的最小复现:第一次抢占成功(受害者进入 2s 冷却),
// 第二次在冷却窗内只能入队——修复前它就永远等不到执行了。
func TestRearm_CooldownExpiryPromotesQueuedInterrupt(t *testing.T) {
a := newPreemptAgent(t, newPreemptProvider())
victim := &Task{ID: 1, Class: TaskInterrupt, Level: LevelBackground, EnqueuedAt: time.Now()}
a.sched.immediate = victim
a.sched.nextRef() // running = victim
// 模拟「刚被抢占过」:冷却起点就在此刻,且抢占提升已生效(有效级 L2
victim.PreemptCount = 1
victim.LastPreemptAt = time.Now()
evt, _ := textEvent("cli", "第二次终止")
evt.Payload["interrupt"] = true
if a.sched.requestPreempt(evt, LevelCritical) {
t.Fatal("抢占冷却期内不得抢占(应入队)")
}
if got := a.sched.stats.PreemptsByLevel[LevelCritical]; got != 0 {
t.Fatalf("被冷却拦成入队的中断不得计入抢占数,实际 %d", got)
}
if n := len(a.sched.interruptQueues[LevelCritical]); n != 1 {
t.Fatalf("应恰好入队一条,实际 %d>1 说明入队路径重复)", n)
}
// 冷却期满 → 安全点的「重新求值」必须把它武装起来(修复前缺失的正是这一步)。
victim.LastPreemptAt = time.Now().Add(-3 * time.Second)
a.sched.rearmPending()
if !a.sched.preemptGrantedFor() {
t.Fatal("冷却期结束后应重新武装让位信号(设计 §4.3「第一个安全点重新求值」)")
}
snap := a.DumpScheduler()
if snap.Immediate == nil {
t.Fatal("重新求值后应把该中断提升进 immediate 槽")
}
if got := snap.Stats.PreemptsByLevel[LevelCritical]; got != 1 {
t.Fatalf("真正占住 immediate 才能计一次抢占,实际 %d", got)
}
// 重新求值不该把任务复制一份队列必须空、immediate 恰好一条。
if n := len(snap.InterruptQueues[LevelCritical]); n != 0 {
t.Fatalf("提升后 L4 队列应空,实际 %d", n)
}
}
// 同一安全点前到达的两条同级中断immediate 是单槽,第二条只能降级入队;
// 它**没有**进入 immediate因此不得计入 PreemptsByLevel也不得被入队两次。
func TestRearm_SameLevelSecondPreempterIsQueuedNotCounted(t *testing.T) {
a := newPreemptAgent(t, newPreemptProvider())
victim := &Task{ID: 1, Class: TaskQueued, EnqueuedAt: time.Now()}
a.sched.immediate = victim
a.sched.nextRef() // running = 排队任务(有效级 0任何中断都能抢
e1, _ := textEvent("cli", "irq-1")
if !a.sched.requestPreempt(e1, LevelInteractive) {
t.Fatal("第一条 L3 应抢占排队任务")
}
e2, _ := textEvent("cli", "irq-2")
a.sched.requestPreempt(e2, LevelInteractive) // 同级 → 降级入队
snap := a.DumpScheduler()
if got := snap.Stats.PreemptsByLevel[LevelInteractive]; got != 1 {
t.Fatalf("被降级的同级第二条不得计入抢占数(期望 1实际 %d", got)
}
if n := len(snap.InterruptQueues[LevelInteractive]); n != 1 {
t.Fatalf("被降级的那条应在 L3 队列里**恰好**出现一次,实际 %d", n)
}
if snap.Immediate == nil {
t.Fatal("第一条应留在 immediate 槽,两条都不能丢")
}
}
// 状态面 Preempted 必须是「各级抢占数之和」,不能拿 Suspended 顶替。
func TestStatus_PreemptedEqualsSumOfLevels(t *testing.T) {
a := New(AgentConfig{ID: "rt-sum", ProviderManager: agentAPI.NewProviderManager(), IO: agentIO.NewIOManager()})
if a.sched == nil {
t.Fatal("agent 应带调度器")
}
a.sched.mu.Lock()
a.sched.stats.PreemptsByLevel[LevelBackground] = 2
a.sched.stats.PreemptsByLevel[LevelInteractive] = 3
a.sched.stats.Suspended = 99 // 故意与抢占数不等
a.sched.mu.Unlock()
got := a.schedulerStatus()
if got.Preempted != 5 {
t.Fatalf("Preempted 应为各级抢占数之和 5实际 %d拿 Suspended 顶替会得 99", got.Preempted)
}
if got.Suspended != 99 {
t.Fatalf("Suspended 应原样透传,实际 %d", got.Suspended)
}
}
// 停机必须给「从未运行」与「已挂起」的同步任务补终态,
// 否则 cli / clawhubadapter 这类无超时的同步注入方会永久挂起(设计 §7 I5、§11.3 X2/X4
func TestStop_DrainsPendingSyncTasks(t *testing.T) {
a := newPreemptAgent(t, newPreemptProvider())
queuedEvt, queuedCh := textEvent("cli", "排队中,永远不会被调度")
if !a.sched.enqueue(newInputTask(queuedEvt)) {
t.Fatal("入队失败")
}
suspEvt, suspCh := textEvent("cli", "已挂起,停机时不会恢复")
a.sched.suspend(
&Task{ID: 2, Class: TaskInterrupt, Level: LevelInteractive, EnqueuedAt: time.Now(), Event: suspEvt},
a.newTaskFrame("挂起", a.stageCtxFromInput("挂起", "", "")),
)
a.Stop()
for name, ch := range map[string]chan *agentIO.OutputEvent{"排队": queuedCh, "挂起": suspCh} {
select {
case r := <-ch:
if r == nil || r.Payload["skipped"] != true {
t.Fatalf("%s 任务停机时应补 skipped 终态,实际 %+v", name, r)
}
case <-time.After(2 * time.Second):
t.Fatalf("%s 任务停机未补终态(同步调用方会永久挂起)", name)
}
}
}
// 背压计数:持续满只报一次「翻转」不发生;计数本身每次都要累加。
func TestBackpressure_CounterAndTransition(t *testing.T) {
s := newScheduler(1)
if !s.noteBackpressure() {
t.Fatal("首次背压应报告「翻转」")
}
if s.noteBackpressure() {
t.Fatal("持续背压不得重复报告翻转(否则日志会被刷爆)")
}
if s.stats.Backpressure != 2 {
t.Fatalf("背压计数应为 2实际 %d", s.stats.Backpressure)
}
s.clearBackpressure()
if !s.noteBackpressure() {
t.Fatal("队列恢复后再满应再次报告翻转")
}
}
// 集成:就绪队列满时 pumpInbox 必须计一次背压Rejected 保持 0——背压不是丢弃
func TestBackpressure_PumpInboxCountsWhenFull(t *testing.T) {
a := newPreemptAgent(t, newPreemptProvider())
a.sched.maxQueue = 1
a.io.InjectInput("cli", "text", map[string]interface{}{"content": "第一条"})
a.io.InjectInput("cli", "text", map[string]interface{}{"content": "第二条"})
a.pumpInbox()
snap := a.DumpScheduler()
if len(snap.Queue) != 1 {
t.Fatalf("maxQueue=1 时队列应恰好 1 条,实际 %d", len(snap.Queue))
}
if snap.Stats.Backpressure == 0 {
t.Fatal("队列满必须计一次背压(设计 §4.4/Q4阻塞发送方 + 计数)")
}
if snap.Stats.Rejected != 0 {
t.Fatalf("背压不是丢弃Rejected 必须保持 0实际 %d", snap.Stats.Rejected)
}
}

View File

@ -178,6 +178,10 @@ func (a *Agent) runTaskSteps(f *TaskFrame) stepOutcome {
for i := 0; i < maxSteps; i++ { for i := 0; i < maxSteps; i++ {
// 安全点:只在 step 之间检查让位。临界区StepToolExec不在此列 // 安全点:只在 step 之间检查让位。临界区StepToolExec不在此列
// 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。 // 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。
//
// 先「重新求值」再判让位:临界区(或抢占冷却期)内被拦成入队的中断,
// 必须在这里重新武装——否则它只能等当前任务自然结束(设计 §4.3/§5.2)。
a.sched.rearmPending()
if !isCriticalChannel(f.OutputChannel) && a.sched.preemptGrantedFor() && a.sched.canSuspend() { if !isCriticalChannel(f.OutputChannel) && a.sched.preemptGrantedFor() && a.sched.canSuspend() {
return outcomeSuspended return outcomeSuspended
} }

View File

@ -411,7 +411,12 @@ func (b *Bridge) readLoop() {
_ = ws.writePong() _ = ws.writePong()
continue continue
} }
// 超时或其他错误,退出 // 超时或其他错误,退出
//
// **必须记日志**:此前这里静默 return设备断线的真因读超时 / 对端
// 关闭 / 帧错)在设备侧完全不可见,只能靠对端日志倒推。
// 2 倍 ping 间隔内的读超时通常是“心跳没人回”——查服务端 writePong 是否真发出。
log.Printf("[devicebridge] read loop exit (opcode=%#x, close=%v): %v", opcode, isClose, err)
return return
} }
if isClose { if isClose {

View File

@ -69,6 +69,65 @@ function sdk.inject_text_no_memory(source, channel, text)
print("[lua-plugin] inject_text_no_memory: " .. tostring(source)) print("[lua-plugin] inject_text_no_memory: " .. tostring(source))
end end
-- !impl
-- opts: { no_memory=bool, context_policy="none"|"prune", cleaner_name=string, priority="L1".."L3" }
-- 零值/缺省 = 记入记忆 + 不裁剪(与三参数版本等价)。
function sdk.inject_text_opts(source, channel, text, opts)
print("[lua-plugin] inject_text_opts: " .. tostring(source))
end
-- !impl
function sdk.inject_interrupt_opts(source, channel, text, opts)
print("[lua-plugin] inject_interrupt_opts: " .. tostring(source))
end
-- !impl
-- 同步注入:等待本轮回复 -> (reply, err);无回复时 reply 为 nil。
function sdk.inject_input_sync(source, channel, text) return nil, nil end
-- !impl
function sdk.inject_input_sync_opts(source, channel, text, opts) return nil, nil end
-- !impl
-- blocks: ContentBlock 数组,见 sdk.inject_input_media。
-- 设置下一轮 tool message 携带的多模态内容块(模型据此看图/听音频)。
function sdk.set_tool_blocks(blocks)
print("[lua-plugin] set_tool_blocks: " .. tostring(blocks and #blocks or 0))
end
-- !impl
-- blocks 每项:{ type="text", text="..." }
-- | { type="image_url", image_url={ url="...", detail="high" } }
-- | { type="audio_url", audio_url={ url="..." } }
function sdk.inject_input_media(source, channel, text, blocks)
print("[lua-plugin] inject_input_media: " .. tostring(source))
end
-- !impl
function sdk.inject_input_media_opts(source, channel, text, blocks, opts)
print("[lua-plugin] inject_input_media_opts: " .. tostring(source))
end
-- !impl
function sdk.inject_input_media_sync(source, channel, text, blocks) return nil, nil end
-- !impl
function sdk.inject_input_media_sync_opts(source, channel, text, blocks, opts) return nil, nil end
-- !impl
function sdk.inject_interrupt_media(source, channel, text, blocks)
print("[lua-plugin] inject_interrupt_media: " .. tostring(source))
end
-- !impl
function sdk.inject_interrupt_media_opts(source, channel, text, blocks, opts)
print("[lua-plugin] inject_interrupt_media_opts: " .. tostring(source))
end
-- !impl
-- 注销输出通道(随资源生灭的动态通道,如远程设备)。返回 (nil, err)。
function sdk.unregister_output_channel(name) return nil, nil end
-- !impl -- !impl
-- enabled: true/false崩溃时内核自动拉起 -- enabled: true/false崩溃时内核自动拉起
function sdk.set_auto_restart(enabled) function sdk.set_auto_restart(enabled)
@ -101,6 +160,9 @@ function sdk.doc.query(text, top_k) return {} end
-- doc: { id=, title=, content= } -- doc: { id=, title=, content= }
function sdk.doc.insert(doc) return nil end function sdk.doc.insert(doc) return nil end
-- !impl -- !impl
-- attachments 每项:{ digest=, mime=, name=, data=<base64> }
function sdk.doc.insert_with_media(doc, attachments) return nil end
-- !impl
function sdk.doc.remove(id) return nil end function sdk.doc.remove(id) return nil end
-- !impl -- !impl
function sdk.doc.stats() return {} end function sdk.doc.stats() return {} end
@ -174,6 +236,24 @@ function sdk.settings.dump() return {} end
-- !impl -- !impl
function sdk.settings.plugins() return {} end function sdk.settings.plugins() return {} end
-- ============ events只读订阅 ============
-- !impl
-- subscribe(event_type, handler) -> unsubscribe()
-- handler 收到 { type=, source=, timestamp=, payload= }
-- 回调在其内核事件发布 goroutine 上执行只做轻量转发不可阻塞Lua 单状态 + 互斥锁)。
sdk.events = {}
function sdk.events.subscribe(event_type, handler)
print("[lua-plugin] events.subscribe: " .. tostring(event_type))
return function() end
end
-- ============ plugin_mgr ============
-- !impl
sdk.plugin_mgr = {}
function sdk.plugin_mgr.reload_one(name) return nil end
function sdk.plugin_mgr.list_loaded() return {} end
function sdk.plugin_mgr.is_disabled(name) return false end
-- json utils (pure Lua) -- json utils (pure Lua)
sdk.json = {} sdk.json = {}

View File

@ -37,7 +37,15 @@ var (
// 1.3.5:系统提示词(人格卡)支持版本占位符 —— 人格卡是配置项,写死版本号 // 1.3.5:系统提示词(人格卡)支持版本占位符 —— 人格卡是配置项,写死版本号
// 会随发版说谎(线上写 v1.0.3、内核 1.3.xagent 就自报 1.0.3)。 // 会随发版说谎(线上写 v1.0.3、内核 1.3.xagent 就自报 1.0.3)。
// 支持 {{kernel_version}} / {{kernel_commit}} / {{sdk_version}}。 // 支持 {{kernel_version}} / {{kernel_commit}} / {{sdk_version}}。
Version = "1.3.10" // 1.3.11Lua 插件桥全量对齐 SDK 1.3.0——把 1.1/1.2/1.3 新增的媒体、
// 注入标志位、中断优先级、事件订阅、动态输出通道注销补进 Lua 侧
// (此前只在 Go 侧存在而文档宣称“完全对齐”)。公开 Go SDK 接口
// 零变更,故 SDK 保持 1.3.0。
// 1.3.12:修 1.3.11 引入的两个真问题 —— ① 驻留子销毁后残留入站 inputch
// child/<id>(改用纯函数名并 Unregister覆盖 destroy/reclaim/StopResidents
// ② sdk.events.subscribe 用了从未注入的公共 Events(),且订阅生命周期
// 管理会自死锁/use-after-close改用内部 Subscribe + 独立 subsMu + Stop 取消)。
Version = "1.3.12"
// Commit 是构建时的 Git commit hash。 // Commit 是构建时的 Git commit hash。
Commit = "unknown" Commit = "unknown"

View File

@ -1,6 +1,7 @@
package plugin package plugin
import ( import (
"encoding/base64"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
@ -9,9 +10,10 @@ import (
"strings" "strings"
"sync" "sync"
lua "github.com/yuin/gopher-lua" agentEvents "gitcode.com/JianFeeeee/HomeAgent/internal/events"
luaSDK "gitcode.com/JianFeeeee/HomeAgent/internal/lua/sdk" luaSDK "gitcode.com/JianFeeeee/HomeAgent/internal/lua/sdk"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
lua "github.com/yuin/gopher-lua"
) )
type toolReg struct { type toolReg struct {
@ -40,7 +42,16 @@ type luaPlugin struct {
stages map[sdk.Stage]*stageReg stages map[sdk.Stage]*stageReg
outputChs map[string]*outputChReg outputChs map[string]*outputChReg
inputDefs map[string]sdk.ChannelDef inputDefs map[string]sdk.ChannelDef
mu sync.Mutex // subs 是本插件注册的事件订阅取消函数Stop 时兜底取消,
// 避免 L 已 Close 后残留回调被触发use-after-close
// 用独立的 subsMu 而非 musubscribe 会在 Lua 的 start 回调里被调,
// 而 Start 正持着 mu —— 用 mu 就是不可重入的自死锁。
subs []func()
subsMu sync.Mutex
// closed 在 Stop 里置位(持 mu事件回调持 mu 后先查它,
// 防止“回调已通过取消订阅检查、但等锁期间 L 被 Close”的竞态。
closed bool
mu sync.Mutex
} }
func newLuaPlugin(luaPath, name string) (*luaPlugin, error) { func newLuaPlugin(luaPath, name string) (*luaPlugin, error) {
@ -287,6 +298,19 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
return 0 return 0
})) }))
// pushReply 统一同步注入的返回约定:非空回复返回 (reply, nil)
// 无回复返回 (nil, nil),与数据类 API 的 (result, err) 约定一致。
pushReply := func(reply string) int {
if reply == "" {
L.Push(lua.LNil)
L.Push(lua.LNil)
return 2
}
L.Push(lua.LString(reply))
L.Push(lua.LNil)
return 2
}
t.RawSetString("inject_text", L.NewFunction(func(L *lua.LState) int { t.RawSetString("inject_text", L.NewFunction(func(L *lua.LState) int {
s.InjectText(L.CheckString(1), L.CheckString(2), L.CheckString(3)) s.InjectText(L.CheckString(1), L.CheckString(2), L.CheckString(3))
return 0 return 0
@ -300,6 +324,67 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
return 0 return 0
})) }))
// ---- 1.2.0 注入标志位no_memory / context_policy / cleaner_name / priority----
t.RawSetString("inject_text_opts", L.NewFunction(func(L *lua.LState) int {
s.InjectTextOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), parseInjectOptions(L, 4))
return 0
}))
t.RawSetString("inject_interrupt_opts", L.NewFunction(func(L *lua.LState) int {
s.InjectInterruptTextOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), parseInjectOptions(L, 4))
return 0
}))
// ---- 同步注入:注入后等待本轮回复,返回 (reply, err) ----
// 注意:内置 SDK 的同名 InjectInputSync 是 (eventType, payload) 形态并遮蔽了
// 公共 SDK 的三参文本版本,故这里显式走 PluginSDK 的公共方法。
t.RawSetString("inject_input_sync", L.NewFunction(func(L *lua.LState) int {
return pushReply(s.PluginSDK.InjectInputSync(L.CheckString(1), L.CheckString(2), L.CheckString(3)))
}))
t.RawSetString("inject_input_sync_opts", L.NewFunction(func(L *lua.LState) int {
return pushReply(s.InjectInputSyncOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), parseInjectOptions(L, 4)))
}))
// ---- 多模态注入1.1.0):内容块随下一次 LLM 请求送达 ----
t.RawSetString("set_tool_blocks", L.NewFunction(func(L *lua.LState) int {
s.SetToolBlocks(luaToContentBlocks(L, 1))
return 0
}))
t.RawSetString("inject_input_media", L.NewFunction(func(L *lua.LState) int {
s.InjectInputMedia(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4))
return 0
}))
t.RawSetString("inject_input_media_opts", L.NewFunction(func(L *lua.LState) int {
s.InjectInputMediaOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4), parseInjectOptions(L, 5))
return 0
}))
t.RawSetString("inject_input_media_sync", L.NewFunction(func(L *lua.LState) int {
return pushReply(s.InjectInputMediaSync(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4)))
}))
t.RawSetString("inject_input_media_sync_opts", L.NewFunction(func(L *lua.LState) int {
return pushReply(s.InjectInputMediaSyncOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4), parseInjectOptions(L, 5)))
}))
t.RawSetString("inject_interrupt_media", L.NewFunction(func(L *lua.LState) int {
s.InjectInterruptMedia(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4))
return 0
}))
t.RawSetString("inject_interrupt_media_opts", L.NewFunction(func(L *lua.LState) int {
s.InjectInterruptMediaOpts(L.CheckString(1), L.CheckString(2), L.CheckString(3), luaToContentBlocks(L, 4), parseInjectOptions(L, 5))
return 0
}))
// ---- 1.3.0 动态输出通道注销:随资源生灭的通道(如远程设备)必须能注销,
// 否则 output_list_channels 会一直列着死通道骗模型。 ----
t.RawSetString("unregister_output_channel", L.NewFunction(func(L *lua.LState) int {
if err := s.UnregisterOutputChannel(L.CheckString(1)); err != nil {
L.Push(lua.LNil)
L.Push(lua.LString(err.Error()))
return 2
}
L.Push(lua.LNil)
L.Push(lua.LNil)
return 2
}))
// ---- 数据类 API与 C ABI 外部插件面完全对齐)---- // ---- 数据类 API与 C ABI 外部插件面完全对齐)----
// 约定:结果型返回 (result, err)void 型返回 (nil, err),成功时 err 为 nil。 // 约定:结果型返回 (result, err)void 型返回 (nil, err),成功时 err 为 nil。
@ -373,13 +458,21 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
if tbl := L.OptTable(1, nil); tbl != nil { if tbl := L.OptTable(1, nil); tbl != nil {
tbl.ForEach(func(_, v lua.LValue) { tbl.ForEach(func(_, v lua.LValue) {
if t2, ok := v.(*lua.LTable); ok { if t2, ok := v.(*lua.LTable); ok {
// sentence_text / media_digests 是媒体绑定链的必经环节:
// 媒体引用挂在句子上,漏掉这两个字段会让图片永远绑不上记忆。
var digests []string
if mt, ok := t2.RawGetString("media_digests").(*lua.LTable); ok {
mt.ForEach(func(_, e lua.LValue) { digests = append(digests, e.String()) })
}
triples = append(triples, sdk.Triple{ triples = append(triples, sdk.Triple{
Subject: t2.RawGetString("subject").String(), Subject: t2.RawGetString("subject").String(),
Relation: t2.RawGetString("relation").String(), Relation: t2.RawGetString("relation").String(),
Object: t2.RawGetString("object").String(), Object: t2.RawGetString("object").String(),
Confidence: float64(lua.LVAsNumber(t2.RawGetString("confidence"))), Confidence: float64(lua.LVAsNumber(t2.RawGetString("confidence"))),
SubjectType: t2.RawGetString("subject_type").String(), SubjectType: t2.RawGetString("subject_type").String(),
ObjectType: t2.RawGetString("object_type").String(), ObjectType: t2.RawGetString("object_type").String(),
SentenceText: t2.RawGetString("sentence_text").String(),
MediaDigests: digests,
}) })
} }
}) })
@ -442,12 +535,17 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
})) }))
docTbl.RawSetString("insert", L.NewFunction(func(L *lua.LState) int { docTbl.RawSetString("insert", L.NewFunction(func(L *lua.LState) int {
if dm := s.DocMemory(); dm != nil { if dm := s.DocMemory(); dm != nil {
tbl := L.CheckTable(1) if err := dm.Insert(docFromLua(L.CheckTable(1))); err != nil {
if err := dm.Insert(&sdk.Doc{ return pushErr(err)
ID: tbl.RawGetString("id").String(), }
Title: tbl.RawGetString("title").String(), }
Content: tbl.RawGetString("content").String(), return pushNil()
}); err != nil { }))
// insert_with_media1.1.0):文档直接持有媒体块,文档向量融合其原生向量,
// 图片按自己的向量被召回,不依赖任何生成的描述文本。
docTbl.RawSetString("insert_with_media", L.NewFunction(func(L *lua.LState) int {
if dm := s.DocMemory(); dm != nil {
if err := dm.InsertWithMedia(docFromLua(L.CheckTable(1)), luaToAttachments(L, L.Get(2))); err != nil {
return pushErr(err) return pushErr(err)
} }
} }
@ -503,10 +601,11 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
if tmem := s.TextMemory(); tmem != nil { if tmem := s.TextMemory(); tmem != nil {
tbl := L.CheckTable(1) tbl := L.CheckTable(1)
if err := tmem.Append(sdk.TextEvent{ if err := tmem.Append(sdk.TextEvent{
Role: tbl.RawGetString("role").String(), Role: tbl.RawGetString("role").String(),
Content: tbl.RawGetString("content").String(), Content: tbl.RawGetString("content").String(),
Timestamp: int64(lua.LVAsNumber(tbl.RawGetString("timestamp"))), Timestamp: int64(lua.LVAsNumber(tbl.RawGetString("timestamp"))),
Channel: tbl.RawGetString("channel").String(), Channel: tbl.RawGetString("channel").String(),
Attachments: luaToAttachments(L, tbl.RawGetString("attachments")),
}); err != nil { }); err != nil {
return pushErr(err) return pushErr(err)
} }
@ -700,6 +799,76 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
} }
return pushVal([]interface{}{}) return pushVal([]interface{}{})
})) }))
// ---- sdk.events.*(只读事件订阅)----
//
// 用内部 SDK 的 Subscribe内置插件用的是同一条路径
// 不用公共 SDK 的 Events()——那个 subscriber 在本内核里从未被注入
// SetEventSubscriber 无调用点),拿到的永远是 nil。
//
// 回调用内核事件发布 goroutine 上执行必须只做轻量转发Lua 单状态 + 互斥锁);
// 阻塞会卡死本插件的全部调用。返回一个取消订阅函数,并在 Stop 时兜底取消
// (否则插件停掉/重载后 L 已 Close残留回调再触发就是 use-after-close
evTbl := subTable("events")
evTbl.RawSetString("subscribe", L.NewFunction(func(L *lua.LState) int {
eventType := L.CheckString(1)
fn := L.CheckFunction(2)
unsub := s.Subscribe(agentEvents.EventType(eventType), func(evt *agentEvents.Event) {
plg.mu.Lock()
defer plg.mu.Unlock()
if plg.closed {
return
}
L2 := plg.L
tbl := L2.NewTable()
tbl.RawSetString("type", lua.LString(string(evt.Type)))
tbl.RawSetString("source", lua.LString(evt.Source))
tbl.RawSetString("timestamp", lua.LNumber(evt.Timestamp))
tbl.RawSetString("payload", goValueToLua(L2, evt.Payload))
L2.Push(fn)
L2.Push(tbl)
if err := L2.PCall(1, 0, nil); err != nil {
fmt.Printf("[lua-plugin/%s] event handler error: %v\n", plg.name, err)
}
})
plg.subsMu.Lock()
plg.subs = append(plg.subs, unsub)
plg.subsMu.Unlock()
L.Push(L.NewFunction(func(L *lua.LState) int {
unsub() // 事件总线的取消订阅是幂等的(重复调用只会匹配不到)
return 0
}))
L.Push(lua.LNil)
return 2
}))
// ---- sdk.plugin_mgr.*(插件管理,与外部插件的 PluginMgrAPI 对齐)----
// PluginMgr 可能未装配(如部分单测的 SDK 构造),此时返回"不可用"而不是 panic。
pmTbl := subTable("plugin_mgr")
pmTbl.RawSetString("reload_one", L.NewFunction(func(L *lua.LState) int {
pm := s.PluginMgr()
if pm == nil {
return pushErr(fmt.Errorf("plugin manager unavailable"))
}
if err := pm.ReloadOne(L.CheckString(1)); err != nil {
return pushErr(err)
}
return pushNil()
}))
pmTbl.RawSetString("list_loaded", L.NewFunction(func(L *lua.LState) int {
pm := s.PluginMgr()
if pm == nil {
return pushList([]interface{}{})
}
return pushList(pm.ListLoadedPlugins())
}))
pmTbl.RawSetString("is_disabled", L.NewFunction(func(L *lua.LState) int {
pm := s.PluginMgr()
if pm == nil {
return pushVal(false)
}
return pushVal(pm.IsPluginDisabled(L.CheckString(1)))
}))
} }
func makeToolHandler(plg *luaPlugin, name string, fn *lua.LFunction) sdk.ToolHandler { func makeToolHandler(plg *luaPlugin, name string, fn *lua.LFunction) sdk.ToolHandler {
@ -724,13 +893,29 @@ func makeStageHandler(plg *luaPlugin, stage sdk.Stage, fn *lua.LFunction) sdk.St
defer plg.mu.Unlock() defer plg.mu.Unlock()
L := plg.L L := plg.L
ctx := map[string]interface{}{ ctx := map[string]interface{}{
"raw_message": sc.RawMessage, "raw_message": sc.RawMessage,
"user_id": sc.UserID, "user_id": sc.UserID,
"group_id": sc.GroupID, "group_id": sc.GroupID,
"phase": string(sc.Phase), "phase": string(sc.Phase),
"llm_text": sc.LLMText, "llm_text": sc.LLMText,
"final_text": sc.FinalText, "reasoning_content": sc.ReasoningContent,
"no_memory": sc.NoMemory, "final_text": sc.FinalText,
"no_memory": sc.NoMemory,
}
if len(sc.ContextMsgs) > 0 {
ctx["context_msgs"] = jsonToIface(sc.ContextMsgs)
}
if len(sc.TokenUsage) > 0 {
ctx["token_usage"] = jsonToIface(sc.TokenUsage)
}
if len(sc.Memory) > 0 {
ctx["memory"] = jsonToIface(sc.Memory)
}
if len(sc.Extra) > 0 {
ctx["extra"] = jsonToIface(sc.Extra)
}
if len(sc.Errors) > 0 {
ctx["errors"] = jsonToIface(sc.Errors)
} }
if sc.Response != nil { if sc.Response != nil {
ctx["response"] = *sc.Response ctx["response"] = *sc.Response
@ -815,6 +1000,7 @@ func parseToolDef(L *lua.LState, defTbl *lua.LTable, plg *luaPlugin, name string
if v := defTbl.RawGetString("no_memory"); v != nil { if v := defTbl.RawGetString("no_memory"); v != nil {
goDef.NoMemory = lua.LVAsBool(v) goDef.NoMemory = lua.LVAsBool(v)
} }
goDef.ContextPolicy = defTbl.RawGetString("context_policy").String()
if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction { if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction {
goDef.Cleaner = makeLuaCleaner(plg, v.(*lua.LFunction)) goDef.Cleaner = makeLuaCleaner(plg, v.(*lua.LFunction))
} }
@ -834,12 +1020,104 @@ func parseChannelDef(L *lua.LState, defTbl *lua.LTable, plg *luaPlugin) sdk.Chan
if v := defTbl.RawGetString("no_memory"); v != nil { if v := defTbl.RawGetString("no_memory"); v != nil {
chDef.NoMemory = lua.LVAsBool(v) chDef.NoMemory = lua.LVAsBool(v)
} }
chDef.ContextPolicy = defTbl.RawGetString("context_policy").String()
if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction { if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction {
chDef.Cleaner = makeLuaCleaner(plg, v.(*lua.LFunction)) chDef.Cleaner = makeLuaCleaner(plg, v.(*lua.LFunction))
} }
return chDef return chDef
} }
// parseInjectOptions 解析 Lua 侧 options table 为 SDK InjectOptions。
// 支持的键no_memory(bool)、context_policy(string)、cleaner_name(string)、priority(string)。
// 缺省/非表等价于零值(记入记忆 + 不裁剪),与旧的三参数注入完全等价。
func parseInjectOptions(L *lua.LState, idx int) sdk.InjectOptions {
opts := sdk.InjectOptions{}
tbl, ok := L.Get(idx).(*lua.LTable)
if !ok {
return opts
}
if v := tbl.RawGetString("no_memory"); v != nil {
opts.NoMemory = lua.LVAsBool(v)
}
opts.ContextPolicy = tbl.RawGetString("context_policy").String()
opts.CleanerName = tbl.RawGetString("cleaner_name").String()
opts.Priority = tbl.RawGetString("priority").String()
return opts
}
// luaToContentBlocks 把 Lua 的 blocks 数组解析为 SDK ContentBlock。
// 每项形如:
//
// { type = "text", text = "..." }
// { type = "image_url", image_url = { url = "...", detail = "high" } }
// { type = "audio_url", audio_url = { url = "..." } }
func luaToContentBlocks(L *lua.LState, idx int) []sdk.ContentBlock {
tbl, ok := L.Get(idx).(*lua.LTable)
if !ok {
return nil
}
var blocks []sdk.ContentBlock
tbl.ForEach(func(_, v lua.LValue) {
bt, ok := v.(*lua.LTable)
if !ok {
return
}
b := sdk.ContentBlock{
Type: bt.RawGetString("type").String(),
Text: bt.RawGetString("text").String(),
}
if iu, ok := bt.RawGetString("image_url").(*lua.LTable); ok {
b.ImageURL = &sdk.ImageURL{
URL: iu.RawGetString("url").String(),
Detail: iu.RawGetString("detail").String(),
}
}
if au, ok := bt.RawGetString("audio_url").(*lua.LTable); ok {
b.AudioURL = &sdk.AudioURL{URL: au.RawGetString("url").String()}
}
blocks = append(blocks, b)
})
return blocks
}
// luaToAttachments 把 Lua 附件数组解析为 SDK MediaAttachment。
// 每项:{ digest=, mime=, name=, data=<base64 字符串> }。
// 带 data 的是新内容(内核落进内容寻址存储),只带 digest 的是引用已有内容。
// base64 解码失败时忽略 data不整单失败——坏附件不应阻断一条记忆写入。
func luaToAttachments(L *lua.LState, val lua.LValue) []sdk.MediaAttachment {
tbl, ok := val.(*lua.LTable)
if !ok {
return nil
}
var out []sdk.MediaAttachment
tbl.ForEach(func(_, v lua.LValue) {
at, ok := v.(*lua.LTable)
if !ok {
return
}
a := sdk.MediaAttachment{
Digest: at.RawGetString("digest").String(),
MIME: at.RawGetString("mime").String(),
Name: at.RawGetString("name").String(),
}
if s := at.RawGetString("data").String(); s != "" {
if b, err := base64.StdEncoding.DecodeString(s); err == nil {
a.Data = b
}
}
out = append(out, a)
})
return out
}
func docFromLua(tbl *lua.LTable) *sdk.Doc {
return &sdk.Doc{
ID: tbl.RawGetString("id").String(),
Title: tbl.RawGetString("title").String(),
Content: tbl.RawGetString("content").String(),
}
}
// jsonToIface 通过 JSON 往返把任意 Go 值转换为 JSON 兼容的 interface{} 树。 // jsonToIface 通过 JSON 往返把任意 Go 值转换为 JSON 兼容的 interface{} 树。
func jsonToIface(v interface{}) interface{} { func jsonToIface(v interface{}) interface{} {
b, err := json.Marshal(v) b, err := json.Marshal(v)
@ -959,8 +1237,21 @@ func (p *luaPlugin) Start(s *sdk.PluginSDK) error {
} }
func (p *luaPlugin) Stop() error { func (p *luaPlugin) Stop() error {
// ① 先取消事件订阅。**不持 p.mu**Bus.Publish 持总线锁回调 handler
// 而 handler 要 p.mu若此处持 p.mu 再取总线锁,就是锁序反转死锁。
p.subsMu.Lock()
subs := p.subs
p.subs = nil
p.subsMu.Unlock()
for _, unsub := range subs {
unsub()
}
// ② 置 closed 并关 L。置位在持锁下完成已进入但等锁的 event 回调
// 拿到锁后会先看到 closed 而直接返回,不会碰已关的 L。
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
p.closed = true
if p.tbl != nil { if p.tbl != nil {
fn := p.tbl.RawGetString("stop") fn := p.tbl.RawGetString("stop")

View File

@ -5,9 +5,10 @@ import (
"path/filepath" "path/filepath"
"testing" "testing"
lua "github.com/yuin/gopher-lua"
internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config" internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config"
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
lua "github.com/yuin/gopher-lua"
) )
func TestTryLoadLua_Basic(t *testing.T) { func TestTryLoadLua_Basic(t *testing.T) {
@ -597,3 +598,66 @@ return plugin
t.Errorf("llm_text writeback: got %q, want %q", sc2.LLMText, "模型输出[尾部标记]") t.Errorf("llm_text writeback: got %q, want %q", sc2.LLMText, "模型输出[尾部标记]")
} }
} }
// TestLuaEventsSubscribeAndStopCleanup 覆盖 sdk.events.subscribe
// 1. 订阅真的能收到内核事件(走内部 SDK 的 Subscribe不是永远为 nil 的公共 Events()
// 2. Stop 会取消订阅,之后 Publish 不得再触碰已 Close 的 LState。
func TestLuaEventsSubscribeAndStopCleanup(t *testing.T) {
dir := t.TempDir()
os.WriteFile(filepath.Join(dir, "plugin.json"), []byte(`{"name":"evlua","entry":"main.lua"}`), 0644)
os.WriteFile(filepath.Join(dir, "main.lua"), []byte(`
local plugin = { name = "evlua" }
function plugin.start(sdk)
_G.hits = 0
local unsub, err = sdk.events.subscribe("agent_output", function(evt)
_G.hits = _G.hits + 1
_G.last_type = evt.type
_G.last_source = evt.source
end)
_G.sub_err = err
_G.unsub_type = type(unsub)
end
function plugin.stop() end
return plugin
`), 0644)
plg, err := tryLoadLua(dir, "evlua", nil)
if err != nil {
t.Fatalf("tryLoadLua failed: %v", err)
}
lp := plg.(*luaPlugin)
bus := events.NewBus()
reg := internalConfig.NewConfigRegistry("")
sett := sdk.NewSettings("evlua", reg)
s := sdk.New("evlua", sdk.SDKConfig{EventBus: bus, Settings: sett})
if err := plg.Start(s); err != nil {
t.Fatalf("Start failed: %v", err)
}
L := lp.L
if errStr := L.GetGlobal("sub_err").String(); errStr != "nil" {
t.Fatalf("subscribe returned error: %s", errStr)
}
if got := L.GetGlobal("unsub_type").String(); got != "function" {
t.Fatalf("subscribe should return an unsubscribe function, got %s", got)
}
bus.Publish(&events.Event{Type: events.EventAgentOutput, Source: "test-src"})
if hits := int(lua.LVAsNumber(L.GetGlobal("hits"))); hits != 1 {
t.Fatalf("event handler hits = %d, want 1", hits)
}
if got := L.GetGlobal("last_source").String(); got != "test-src" {
t.Fatalf("event source = %q, want test-src", got)
}
// Stop 取消订阅 + 关 L此后再 Publish 不得 panic / use-after-close。
if err := plg.Stop(); err != nil {
t.Fatalf("Stop failed: %v", err)
}
bus.Publish(&events.Event{Type: events.EventAgentOutput, Source: "after-stop"})
}

View File

@ -0,0 +1,188 @@
package plugin
import (
"os"
"path/filepath"
"regexp"
"strings"
"testing"
luaSDK "gitcode.com/JianFeeeee/HomeAgent/internal/lua/sdk"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
lua "github.com/yuin/gopher-lua"
)
// TestLuaSDKMockSingleSource 守住「Lua mock 只有一份事实源」。
//
// 三份 sdk.lua内核内嵌 / 工具链模板 / 项目副本)历史上各自漂移过,
// 表现为「mock 里有的 API内核运行时是 nil」这类静默失配。
// 事实源是 SDK 仓的 sdk/lua/sdk.lua内核副本由
// third_party/homeagent-sdk/scripts/sync-lua-sdk.sh 同步。
func TestLuaSDKMockSingleSource(t *testing.T) {
canonical, err := os.ReadFile(filepath.Join("..", "..", "third_party", "homeagent-sdk", "sdk", "lua", "sdk.lua"))
if err != nil {
t.Skipf("SDK repo canonical sdk.lua not available: %v", err)
}
if string(canonical) != luaSDK.SDKSource {
t.Fatal("内核内嵌 sdk.lua 与 SDK 仓 sdk/lua/sdk.lua 不一致;" +
"请跑 third_party/homeagent-sdk/scripts/sync-lua-sdk.sh")
}
}
var luaMockFuncRe = regexp.MustCompile(`(?m)^function sdk\.([A-Za-z0-9_.]+)\s*\(`)
// TestLuaBridgeCoversMock 守住「mock 承诺的每个函数,运行时都有绑定」。
//
// 只查 mock → 运行时这一向mock 定义了但没绑定,插件会先看到 mock 能调、
// 之后内核里是 nil或反向的假象。反向运行时多出未文档化的函数无害。
func TestLuaBridgeCoversMock(t *testing.T) {
bridge, err := os.ReadFile("lua_plugin.go")
if err != nil {
t.Fatalf("read lua_plugin.go: %v", err)
}
src := string(bridge)
// 纯 Lua 实现,不经内核绑定。
exempt := map[string]bool{"json.encode": true, "json.decode": true}
matches := luaMockFuncRe.FindAllStringSubmatch(luaSDK.SDKSource, -1)
if len(matches) < 40 {
t.Fatalf("parsed only %d sdk.* functions from mock; parser likely broken", len(matches))
}
for _, m := range matches {
full := m[1]
if exempt[full] {
continue
}
leaf := full
if i := strings.LastIndex(full, "."); i >= 0 {
leaf = full[i+1:]
}
if !strings.Contains(src, `RawSetString("`+leaf+`"`) {
t.Errorf("sdk.%s: mock 有定义,但 lua_plugin.go 没有 RawSetString(%q) 绑定", full, leaf)
}
}
}
func luaTableFrom(t *testing.T, code string) (*lua.LState, *lua.LTable) {
t.Helper()
L := lua.NewState()
if err := L.DoString("return " + code); err != nil {
L.Close()
t.Fatalf("eval lua table: %v", err)
}
tbl, ok := L.Get(-1).(*lua.LTable)
if !ok {
L.Close()
t.Fatalf("expected a table from %q", code)
}
L.Pop(1)
return L, tbl
}
func TestLuaParseInjectOptions(t *testing.T) {
L, tbl := luaTableFrom(t, `{
no_memory = true,
context_policy = "prune",
cleaner_name = "sanitize",
priority = "L2",
}`)
defer L.Close()
L.Push(tbl)
got := parseInjectOptions(L, 1)
L.Pop(1)
if !got.NoMemory {
t.Error("NoMemory should be true")
}
if got.ContextPolicy != sdk.ContextPolicyPrune {
t.Errorf("ContextPolicy = %q, want prune", got.ContextPolicy)
}
if got.CleanerName != "sanitize" {
t.Errorf("CleanerName = %q, want sanitize", got.CleanerName)
}
if got.Priority != sdk.PriorityL2 {
t.Errorf("Priority = %q, want L2", got.Priority)
}
// 缺省/非表 = 零值(记入记忆 + 不裁剪),与旧三参数注入等价。
if z := parseInjectOptions(L, 99); z != (sdk.InjectOptions{}) {
t.Errorf("missing opts should be zero value, got %#v", z)
}
}
func TestLuaContentBlocksParse(t *testing.T) {
L, tbl := luaTableFrom(t, `{
{ type = "text", text = "看图" },
{ type = "image_url", image_url = { url = "data:image/png;base64,AAAA", detail = "high" } },
{ type = "audio_url", audio_url = { url = "https://x/a.mp3" } },
}`)
defer L.Close()
L.Push(tbl)
blocks := luaToContentBlocks(L, 1)
L.Pop(1)
if len(blocks) != 3 {
t.Fatalf("got %d blocks, want 3", len(blocks))
}
if blocks[0].Type != "text" || blocks[0].Text != "看图" {
t.Errorf("block[0] = %#v", blocks[0])
}
if blocks[1].ImageURL == nil || blocks[1].ImageURL.URL != "data:image/png;base64,AAAA" || blocks[1].ImageURL.Detail != "high" {
t.Errorf("block[1] image_url = %#v", blocks[1].ImageURL)
}
if blocks[2].AudioURL == nil || blocks[2].AudioURL.URL != "https://x/a.mp3" {
t.Errorf("block[2] audio_url = %#v", blocks[2].AudioURL)
}
}
func TestLuaAttachmentsBase64(t *testing.T) {
// "hello" 的 base64 是 aGVsbG8=
L, tbl := luaTableFrom(t, `{
{ digest = "sha256:abc", mime = "image/png", name = "a.png" },
{ mime = "image/jpeg", data = "aGVsbG8=" },
{ mime = "image/png", data = "!!!not-base64!!!" },
}`)
defer L.Close()
atts := luaToAttachments(L, tbl)
if len(atts) != 3 {
t.Fatalf("got %d attachments, want 3", len(atts))
}
if atts[0].Digest != "sha256:abc" || atts[0].MIME != "image/png" || atts[0].Name != "a.png" || atts[0].Data != nil {
t.Errorf("att[0] = %#v", atts[0])
}
if string(atts[1].Data) != "hello" {
t.Errorf("att[1] data = %q, want hello", string(atts[1].Data))
}
// 坏 base64 只丢 data不整单失败——坏附件不应阻断记忆写入。
if atts[2].Data != nil {
t.Errorf("att[2] bad base64 should be dropped, got %q", string(atts[2].Data))
}
}
func TestLuaDefinitionsCarryContextPolicy(t *testing.T) {
L, tbl := luaTableFrom(t, `{
description = "t",
no_memory = true,
context_policy = "prune",
}`)
defer L.Close()
plg := &luaPlugin{name: "cp"}
def := parseToolDef(L, tbl, plg, "t")
if def.ContextPolicy != sdk.ContextPolicyPrune {
t.Errorf("ToolDef.ContextPolicy = %q, want prune", def.ContextPolicy)
}
if !def.NoMemory {
t.Error("ToolDef.NoMemory should be true")
}
chDef := parseChannelDef(L, tbl, plg)
if chDef.ContextPolicy != sdk.ContextPolicyPrune {
t.Errorf("ChannelDef.ContextPolicy = %q, want prune", chDef.ContextPolicy)
}
}

View File

@ -15,7 +15,7 @@ import (
type PluginType string type PluginType string
const ( const (
PluginTypeSKILL PluginType = "skill" PluginTypeSKILL PluginType = "skill"
) )
type IOConfig struct { type IOConfig struct {
@ -78,18 +78,28 @@ func LoadSKILL(path string) (*SKILLPlugin, error) {
metaFile := filepath.Join(path, "skill.json") metaFile := filepath.Join(path, "skill.json")
if data, err := os.ReadFile(metaFile); err == nil { if data, err := os.ReadFile(metaFile); err == nil {
var meta struct { var meta struct {
Name string `json:"name"` Name string `json:"name"`
Description string `json:"description"` Description string `json:"description"`
Version string `json:"version"` Version string `json:"version"`
Author string `json:"author"` Author string `json:"author"`
IO *IOConfig `json:"io,omitempty"` IO *IOConfig `json:"io,omitempty"`
} }
if err := json.Unmarshal(data, &meta); err == nil { if err := json.Unmarshal(data, &meta); err == nil {
if meta.Name != "" { p.name = meta.Name } if meta.Name != "" {
if meta.Description != "" { p.description = meta.Description } p.name = meta.Name
if meta.Version != "" { p.version = meta.Version } }
if meta.Author != "" { p.author = meta.Author } if meta.Description != "" {
if meta.IO != nil { p.ioConfig = meta.IO } p.description = meta.Description
}
if meta.Version != "" {
p.version = meta.Version
}
if meta.Author != "" {
p.author = meta.Author
}
if meta.IO != nil {
p.ioConfig = meta.IO
}
} }
} }
} else if filepath.Ext(path) == ".md" { } else if filepath.Ext(path) == ".md" {
@ -198,7 +208,9 @@ func extractToolDefs(content string) []ToolDef {
inCodeBlock = !inCodeBlock inCodeBlock = !inCodeBlock
continue continue
} }
if inCodeBlock { continue } if inCodeBlock {
continue
}
if strings.HasPrefix(trimmed, "## ") && !strings.HasPrefix(trimmed, "### ") { if strings.HasPrefix(trimmed, "## ") && !strings.HasPrefix(trimmed, "### ") {
if currentTool != nil && currentTool.Name != "" { if currentTool != nil && currentTool.Name != "" {
@ -232,7 +244,9 @@ func extractToolDefs(content string) []ToolDef {
continue continue
} }
if currentTool == nil || currentTool.Name == "" { continue } if currentTool == nil || currentTool.Name == "" {
continue
}
if currentTool.Description == "" && trimmed != "" && if currentTool.Description == "" && trimmed != "" &&
!strings.HasPrefix(trimmed, "- ") && !strings.HasPrefix(trimmed, "#") { !strings.HasPrefix(trimmed, "- ") && !strings.HasPrefix(trimmed, "#") {
@ -280,7 +294,9 @@ func isNonToolSection(name string) bool {
func extractIOConfig(content string) *IOConfig { func extractIOConfig(content string) *IOConfig {
ioType := extractField(content, "io_type") ioType := extractField(content, "io_type")
if ioType == "" { return nil } if ioType == "" {
return nil
}
cfg := &IOConfig{ cfg := &IOConfig{
Type: ioType, Type: ioType,
InputRoute: extractField(content, "io_input_route"), InputRoute: extractField(content, "io_input_route"),

View File

@ -0,0 +1,53 @@
package remotedevice
import (
"net/http"
"net/http/httptest"
"testing"
"time"
)
// 心跳回包必须**真的发出去**pong 只有两个字节,且设备空闲时没有任何别的写
// 会顺带把 bufio 缓冲刷出去——`writePong` 一旦忘了 Flushpong 就永远留在
// 服务端缓冲里。
//
// 这就是「device channel 不稳定」的真因(实测):客户端每 30s 发一个 ping
// 服务端算好了 pong 却没发;客户端的读循环设的是 2 倍 ping 间隔(默认 60s
// 读超时,于是**每 60 秒准点断开一次**,重连后 outputch 被注销又注册,
// 模型侧看到的就是工具/通道凭空消失又出现。
//
// 本用例只发一个 ping随后**什么都不发**pong 必须在无后续流量的情况下到达。
func TestWSPingGetsPongWhileIdle(t *testing.T) {
reg := NewRegistry()
token := "test-token-ping"
reg.SetAcceptToken(func(provided string) bool { return provided == token })
srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS))
defer srv.Close()
cli := dialTestWS(t, srv.URL, token)
defer cli.close()
// 先走完 hello + bind服务端要先把设备登记进 connspong 才写得回来)。
cli.sendText([]byte(`{"op":"hello","device":{"device_id":"ping-dev","name":"前端机","kind":"computer","caps":["cmd"]}}`))
cli.readHelloAckAndBind(t, token)
cli.sendFrame(0x9, nil) // ping
if err := cli.conn.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
t.Fatalf("set read deadline: %v", err)
}
payload, isClose, opcode, err := readFrame(cli.rw.Reader)
if err != nil {
t.Fatalf("2s 内没收到 pongwritePong 忘了 Flush: %v", err)
}
if isClose {
t.Fatal("连接被关闭,而不是回了 pong")
}
if opcode != 0xa {
t.Fatalf("期望 pong(0xa),实际 opcode=%#x payload=%q", opcode, payload)
}
if len(payload) != 0 {
t.Fatalf("pong 不该带负载,实际 %q", payload)
}
}

View File

@ -606,7 +606,13 @@ func writeFrame(w *bufio.Writer, opcode byte, payload []byte) error {
} }
func writePong(w *bufio.Writer) error { func writePong(w *bufio.Writer) error {
return writeFrameHeader(w, 0xa, 0) // 必须走 writeFrame它 Flush
//
// 回归的 bug这里原先是裸的 writeFrameHeader**不 Flush**。设备空闲时
// 没有任何别的写会顺带把 bufio 缓冲刷出去,于是 pong 永远留在服务端缓冲里,
// 客户端等 2 倍 ping 间隔(默认 30s×2 = 60s读超时断开、重连——
// 实测表现就是「设备通道每 60 秒掉线一次」,连带着 outputch 反复注销/注册。
return writeFrame(w, 0xa, nil)
} }
func writeFrameHeader(w *bufio.Writer, opcode byte, length int) error { func writeFrameHeader(w *bufio.Writer, opcode byte, length int) error {

View File

@ -68,7 +68,11 @@ type SchedulerStatus struct {
Rejected uint64 `json:"rejected"` Rejected uint64 `json:"rejected"`
Suspended uint64 `json:"suspended"` Suspended uint64 `json:"suspended"`
Resumed uint64 `json:"resumed"` Resumed uint64 `json:"resumed"`
// Preempted = Σ PreemptsByLevel[1..4],即「真正抢占成功」的次数。
// 它与 Suspended 不等价(受害者可能先自行结束),因此不是 Suspended 的别名。
Preempted uint64 `json:"preempted"` Preempted uint64 `json:"preempted"`
// Backpressure 是就绪队列满、输入被挡回 channel 的次数(暂时不收,不是丢弃)。
Backpressure uint64 `json:"backpressure"`
} }
// SchedulerTask 是任务的最小标识(不暴露帧内容)。 // SchedulerTask 是任务的最小标识(不暴露帧内容)。

View File

@ -0,0 +1,403 @@
-- HomeAgent Lua Plugin SDK
-- Interface contract between Lua plugins and HomeAgent kernel.
-- !impl functions are replaced by Go implementations at runtime.
-- Standalone/debug: pure Lua mock implementations are used.
-- Usage: local sdk = require("sdk")
sdk = {}
-- !impl
-- level: "debug" | "info" | "warn" | "error"
function sdk.log(level, msg)
print("[lua-plugin] " .. tostring(level) .. ": " .. tostring(msg))
end
-- !impl
-- def: { description="...", parameters={...}, no_memory=true/false, cleaner=function(text)->text }
-- handler: function(args) -> result
function sdk.register_tool(name, def, handler)
print("[lua-plugin] register_tool: " .. tostring(name))
end
-- !impl
-- stage: "on_input" | "pre_action" | "post_action" | ...
-- scope: nil/"global" (默认) | "own_tools"(仅 before_toolcall/after_toolcall 且工具属于本插件时触发)
function sdk.register_stage(stage, handler, scope)
print("[lua-plugin] register_stage: " .. tostring(stage) .. " scope=" .. tostring(scope))
end
-- !impl
function sdk.register_api(name)
print("[lua-plugin] register_api: " .. tostring(name))
end
-- !impl
-- def: { no_memory=true/false, cleaner=function(text)->text }
-- handler: function(args) -> result
function sdk.register_output_channel(name, caps, desc, def, handler)
print("[lua-plugin] register_output_channel: " .. tostring(name))
end
-- !impl
-- def: { no_memory=true/false, cleaner=function(text)->text }
function sdk.register_input_channel(name, def)
print("[lua-plugin] register_input_channel: " .. tostring(name))
end
-- !impl
function sdk.get_setting(key)
return nil
end
-- !impl
function sdk.set_setting(key, value)
print("[lua-plugin] set_setting: " .. tostring(key))
end
-- !impl
function sdk.inject_text(source, channel, text)
print("[lua-plugin] inject_text: " .. tostring(source) .. "/" .. tostring(channel))
end
-- !impl
function sdk.inject_interrupt(source, channel, text)
print("[lua-plugin] inject_interrupt: " .. tostring(source))
end
-- !impl
function sdk.inject_text_no_memory(source, channel, text)
print("[lua-plugin] inject_text_no_memory: " .. tostring(source))
end
-- !impl
-- opts: { no_memory=bool, context_policy="none"|"prune", cleaner_name=string, priority="L1".."L3" }
-- 零值/缺省 = 记入记忆 + 不裁剪(与三参数版本等价)。
function sdk.inject_text_opts(source, channel, text, opts)
print("[lua-plugin] inject_text_opts: " .. tostring(source))
end
-- !impl
function sdk.inject_interrupt_opts(source, channel, text, opts)
print("[lua-plugin] inject_interrupt_opts: " .. tostring(source))
end
-- !impl
-- 同步注入:等待本轮回复 -> (reply, err);无回复时 reply 为 nil。
function sdk.inject_input_sync(source, channel, text) return nil, nil end
-- !impl
function sdk.inject_input_sync_opts(source, channel, text, opts) return nil, nil end
-- !impl
-- blocks: ContentBlock 数组,见 sdk.inject_input_media。
-- 设置下一轮 tool message 携带的多模态内容块(模型据此看图/听音频)。
function sdk.set_tool_blocks(blocks)
print("[lua-plugin] set_tool_blocks: " .. tostring(blocks and #blocks or 0))
end
-- !impl
-- blocks 每项:{ type="text", text="..." }
-- | { type="image_url", image_url={ url="...", detail="high" } }
-- | { type="audio_url", audio_url={ url="..." } }
function sdk.inject_input_media(source, channel, text, blocks)
print("[lua-plugin] inject_input_media: " .. tostring(source))
end
-- !impl
function sdk.inject_input_media_opts(source, channel, text, blocks, opts)
print("[lua-plugin] inject_input_media_opts: " .. tostring(source))
end
-- !impl
function sdk.inject_input_media_sync(source, channel, text, blocks) return nil, nil end
-- !impl
function sdk.inject_input_media_sync_opts(source, channel, text, blocks, opts) return nil, nil end
-- !impl
function sdk.inject_interrupt_media(source, channel, text, blocks)
print("[lua-plugin] inject_interrupt_media: " .. tostring(source))
end
-- !impl
function sdk.inject_interrupt_media_opts(source, channel, text, blocks, opts)
print("[lua-plugin] inject_interrupt_media_opts: " .. tostring(source))
end
-- !impl
-- 注销输出通道(随资源生灭的动态通道,如远程设备)。返回 (nil, err)。
function sdk.unregister_output_channel(name) return nil, nil end
-- !impl
-- enabled: true/false崩溃时内核自动拉起
function sdk.set_auto_restart(enabled)
print("[lua-plugin] set_auto_restart: " .. tostring(enabled))
end
-- ============ graph memory ============
-- !impl
sdk.memory = {}
-- !impl
-- query: string, depth: number -> {entities={...}, relations={...}}
function sdk.memory.recall(query, depth) return {entities={}, relations={}} end
-- !impl
-- triples: { {subject=, relation=, object=, [confidence=], [sentence_text=]} } -> err
function sdk.memory.commit(triples) return nil end
-- !impl
function sdk.memory.introspect() return {} end
-- !impl
function sdk.memory.merge(source, target) return 0 end
-- !impl
-- criteria: {key=value}, hard: boolean
function sdk.memory.purge(criteria, hard) return 0 end
-- ============ document memory ============
-- !impl
sdk.doc = {}
-- !impl
function sdk.doc.query(text, top_k) return {} end
-- !impl
-- doc: { id=, title=, content= }
function sdk.doc.insert(doc) return nil end
-- !impl
-- attachments 每项:{ digest=, mime=, name=, data=<base64> }
function sdk.doc.insert_with_media(doc, attachments) return nil end
-- !impl
function sdk.doc.remove(id) return nil end
-- !impl
function sdk.doc.stats() return {} end
-- ============ knowledge ============
-- !impl
sdk.knowledge = {}
-- !impl
function sdk.knowledge.search(query, limit) return {} end
-- !impl
function sdk.knowledge.add(tag, content) return nil end
-- !impl
function sdk.knowledge.list() return {} end
-- ============ text memory ============
-- !impl
sdk.text_memory = {}
-- !impl
-- evt: { timestamp=, role=, content=, channel= }
function sdk.text_memory.append(evt) return nil end
-- ============ llm ============
-- !impl
sdk.llm = {}
-- !impl
function sdk.llm.list_sources() return {} end
-- !impl
function sdk.llm.set_source(name) return nil end
-- !impl
function sdk.llm.current_source() return nil end
-- ============ social (只读) ============
-- !impl
sdk.social = {}
-- !impl
function sdk.social.get_person(name) return {} end
-- !impl
function sdk.social.get_network(name, depth) return {} end
-- !impl
function sdk.social.get_trait(name, trait) return {value=nil, found=false} end
-- !impl
function sdk.social.get_relations(name) return {} end
-- !impl
function sdk.social.list_persons() return {} end
-- ============ settings (作用域变体) ============
-- !impl
sdk.settings = {}
-- !impl
function sdk.settings.get_core(key) return nil end
-- !impl
function sdk.settings.set_core(key, value) return nil end
-- !impl
function sdk.settings.list_core(prefix) return {} end
-- !impl
function sdk.settings.get_plugin(plugin, key) return nil end
-- !impl
function sdk.settings.set_plugin(plugin, key, value) return nil end
-- !impl
function sdk.settings.list_plugin(plugin, prefix) return {} end
-- !impl
function sdk.settings.list(prefix) return {} end
-- !impl
-- def: { key=, type=, display_name=, description=, category=, options=, default=,
-- min=, max=, step=, required=, secret= }
function sdk.settings.register_def(def) return nil end
-- !impl
function sdk.settings.defs(prefix) return {} end
-- !impl
function sdk.settings.dump() return {} end
-- !impl
function sdk.settings.plugins() return {} end
-- ============ events只读订阅 ============
-- !impl
-- subscribe(event_type, handler) -> unsubscribe()
-- handler 收到 { type=, source=, timestamp=, payload= }
-- 回调在其内核事件发布 goroutine 上执行只做轻量转发不可阻塞Lua 单状态 + 互斥锁)。
sdk.events = {}
function sdk.events.subscribe(event_type, handler)
print("[lua-plugin] events.subscribe: " .. tostring(event_type))
return function() end
end
-- ============ plugin_mgr ============
-- !impl
sdk.plugin_mgr = {}
function sdk.plugin_mgr.reload_one(name) return nil end
function sdk.plugin_mgr.list_loaded() return {} end
function sdk.plugin_mgr.is_disabled(name) return false end
-- json utils (pure Lua)
sdk.json = {}
function sdk.json.encode(val)
local ok, result = pcall(function()
local function _encode(v)
local t = type(v)
if t == "string" then
local s = v:gsub('\\', '\\\\'):gsub('"', '\\"'):gsub('\n', '\\n'):gsub('\r', '\\r'):gsub('\t', '\\t')
return '"' .. s .. '"'
elseif t == "number" then
return tostring(v)
elseif t == "boolean" then
return tostring(v)
elseif t == "table" then
local keys = {}
local is_array = true
local maxn = 0
for k in pairs(v) do
keys[#keys + 1] = k
if type(k) ~= "number" or k < 1 or k ~= math.floor(k) then
is_array = false
end
if type(k) == "number" and k > maxn then maxn = k end
end
if is_array and #keys >= maxn then
local parts = {}
for i = 1, maxn do
parts[#parts + 1] = _encode(v[i])
end
return "[" .. table.concat(parts, ",") .. "]"
else
local parts = {}
for _, k in ipairs(keys) do
parts[#parts + 1] = _encode(tostring(k)) .. ":" .. _encode(v[k])
end
return "{" .. table.concat(parts, ",") .. "}"
end
else
return "null"
end
end
return _encode(val)
end)
if ok then return result end
return "null"
end
function sdk.json.decode(str)
local ok, result = pcall(function()
local pos, _end = 1, #str
local function skip()
while pos <= _end and str:sub(pos, pos):match("%s") do pos = pos + 1 end
end
local function parse()
skip()
if pos > _end then return nil end
local c = str:sub(pos, pos)
if c == '"' then
local s = {}
pos = pos + 1
while pos <= _end do
local ch = str:sub(pos, pos)
if ch == '"' then
pos = pos + 1
return table.concat(s)
elseif ch == '\\' then
pos = pos + 1
local n = str:sub(pos, pos)
if n == '"' then s[#s+1] = '"'
elseif n == '\\' then s[#s+1] = '\\'
elseif n == '/' then s[#s+1] = '/'
elseif n == 'b' then s[#s+1] = '\b'
elseif n == 'f' then s[#s+1] = '\f'
elseif n == 'n' then s[#s+1] = '\n'
elseif n == 'r' then s[#s+1] = '\r'
elseif n == 't' then s[#s+1] = '\t'
elseif n == 'u' then
local hex = str:sub(pos+1, pos+4)
pos = pos + 4
s[#s+1] = utf8 and utf8.char(tonumber(hex, 16)) or '?'
end
pos = pos + 1
else
s[#s+1] = ch
pos = pos + 1
end
end
return table.concat(s)
elseif c == 't' then pos = pos + 4; return true
elseif c == 'f' then pos = pos + 5; return false
elseif c == 'n' then pos = pos + 4; return nil
elseif c == '{' then
pos = pos + 1; skip()
local t = {}
if str:sub(pos, pos) == '}' then pos = pos + 1; return t end
while true do
skip(); local k = parse(); skip()
if str:sub(pos, pos) == ':' then pos = pos + 1 end
skip(); t[k] = parse(); skip()
local sep = str:sub(pos, pos)
if sep == '}' then pos = pos + 1; return t end
if sep == ',' then pos = pos + 1 end
end
elseif c == '[' then
pos = pos + 1; skip()
local t = {}
if str:sub(pos, pos) == ']' then pos = pos + 1; return t end
local idx = 1
while true do
skip(); t[idx] = parse(); idx = idx + 1; skip()
local sep = str:sub(pos, pos)
if sep == ']' then pos = pos + 1; return t end
if sep == ',' then pos = pos + 1 end
end
else
local s, e = str:find('^[-%d%.eE]+', pos)
if s then
local num = tonumber(str:sub(s, e))
pos = e + 1
return num
end
return nil
end
end
return parse()
end)
if ok then return result end
return nil
end
-- http utils
sdk.http = {}
-- !impl
function sdk.http.get(url)
print("[lua-plugin] http.get: " .. tostring(url))
return {status=200, body='{"mock":true}', headers={}}
end
-- !impl
function sdk.http.post(url, body, content_type)
print("[lua-plugin] http.post: " .. tostring(url))
return {status=200, body='{"mock":true}', headers={}}
end
return sdk