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` 失败（与生产症状一致）。

.gitignore

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

assets/docs/en/PLUGIN_DEV.md

@@ -599,23 +599,26 @@ When running inside the kernel, `sdk.*` global variables are injected by the Go
 
 ### 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.1–1.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**
 
 | Function | Description |
 |----------|-------------|
 | `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_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.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 |
 
 **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):
 
@@ -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_interrupt(source, channel, text)` | Interrupt delivery |
 | `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)`)**
 
 | Sub-table | Functions |
 |-----------|-----------|
-| `sdk.memory.*` | `recall(query, depth)`, `commit({triples})`, `introspect()`, `merge(source, target)`, `purge(criteria, hard)` |
-| `sdk.doc.*` | `query(text, top_k)`, `insert({id,title,content})`, `remove(id)`, `stats()` |
+| `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})`, `insert_with_media(doc, attachments)`, `remove(id)`, `stats()` |
 | `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.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.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"> :

assets/docs/zh/PLUGIN_DEV.md

@@ -592,23 +592,26 @@ lua main.lua
 
 ### 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.1–1.3 的媒体/注入标志位/优先级能力曾长期只在 Go 侧，Lua 侧静默缺失。现已全量对齐，并由 `internal/plugin/lua_surface_test.go` 的契约测试守住「mock 承诺的每个函数都有运行时绑定」。
 
 **注册类**
 
 | 函数 | 说明 |
 |------|------|
 | `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_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.unregister_output_channel(name)` | 注销输出通道（随资源生灭的动态通道，如远程设备）；返回 `(nil, err)` |
 | `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`（与子进程外部插件能力对齐）：
 
@@ -637,20 +640,37 @@ end)
 | `sdk.inject_text(source, channel, text)` | 投递文本消息 |
 | `sdk.inject_interrupt(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)`）**
 
 | 子表 | 函数 |
 |------|------|
-| `sdk.memory.*` | `recall(query, depth)`、`commit({triples})`、`introspect()`、`merge(source, target)`、`purge(criteria, hard)` |
-| `sdk.doc.*` | `query(text, top_k)`、`insert({id,title,content})`、`remove(id)`、`stats()` |
+| `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})`、`insert_with_media(doc, attachments)`、`remove(id)`、`stats()` |
 | `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.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.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"> :

internal/agent/core/agent.go

@@ -404,9 +404,30 @@ func (a *Agent) Start() {
 func (a *Agent) Stop() {
 	// 父退出**必须**销毁全部驻留子（设计 §10 硬约束：子不得比父活得久、不留孤儿）。
 	a.StopResidents()
+	// 停机前给待办任务补终态。运行中的任务会经 cancel → LLM 失败 → emitResponse
+	// 自然拿到终态，但**从未运行**（排队/待处理）与**已挂起**的任务不会有任何人
+	// 回它们；带 ResponseCh 的同步注入方（cli / clawhubadapter 均无超时）会永久挂起
+	// （设计 §7 I5、§11.3 X2/X4）。必须在 cancel 之前做：cancel 会让调度器直接 return。
+	a.drainPendingInterrupts("agent_stopped")
 	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 的图记忆共同面实现。
 //
 //   - 轻量内核（给了 LightMemory）：用 LightMemory，**整理面保持 nil**；

internal/agent/core/eventloop.go

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

internal/agent/core/output.go

@@ -13,8 +13,13 @@ func (a *Agent) executeOutputSendTool(tc agentAPI.ToolCall) string {
 	channel := strings.TrimPrefix(tc.Name, "output_send__")
 	payload, _ := tc.Arguments["payload"].(string)
 	rawType, _ := tc.Arguments["type"].(string)
-	if channel == "" || payload == "" || rawType == "" {
-		return "工具名称格式: output_send__{channel}，payload 和 type 不能为空"
+	if channel == "" || payload == "" {
+		return "工具名称格式: output_send__{channel}，payload 不能为空"
+	}
+	// type 缺省按 text 处理：绝大多数输出就是文本，让模型为"省略一个默认值"付一次
+	// 失败重试没有意义（判据该拦的是"不知道发什么"，不是"没写众所周知的默认值"）。
+	if rawType == "" {
+		rawType = "text"
 	}
 	// 授权闸（纵深防御）：模型可能凭名字直接调未授权的输出门。
 	if !a.IsOutputAllowed(channel) {

internal/agent/core/output_rules_test.go

@@ -0,0 +1,100 @@
+package core
+
+import (
+	"strings"
+	"testing"
+
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+)
+
+// 设计口径：输出是 agent 的**主动调用** —— 收到一次输入后，可以往任意（已授权的）
+// 通道发**任意多次**（分段播报、先回执后结论、同时通知多个通道都合法）。
+//
+// 这条判据钉住的是"提示词里不得出现输出次数限制"。此前 `tooldefs.go` 里写着
+// 「每轮对话通常只需调用一次 output_send__{通道名} 即可完成回复」—— 一条凭空的限制，
+// 会让模型自己收起合理的多次输出（用户现场指出）。
+func TestSystemPromptDoesNotRestrictOutputCount(t *testing.T) {
+	parent, _, _ := newRootForResidents(t)
+	defer parent.Stop()
+
+	prompt := parent.buildSystemPrompt("", "你好")
+	banned := []string{
+		"只需调用一次",
+		"只能调用一次",
+		"通常只需调用",
+		"不要重复发送",
+	}
+	for _, b := range banned {
+		if strings.Contains(prompt, b) {
+			t.Fatalf("系统提示词里仍有输出次数限制 %q —— 设计上次数不限", b)
+		}
+	}
+	if !strings.Contains(prompt, "输出次数与目标通道由你自己决定") {
+		t.Fatal("系统提示词应明确「输出次数与目标通道由你自己决定」")
+	}
+	if !strings.Contains(prompt, "没有任何「一轮只能发一次」的限制") {
+		t.Fatal("系统提示词应显式否认「一轮只能发一次」")
+	}
+}
+
+// 输出工具的 type 可省略，缺省按 text 处理（判据该拦的是"不知道发什么"，
+// 不是"没写众所周知的默认值"）。
+func TestOutputSendTypeDefaultsToText(t *testing.T) {
+	parent, _, _ := newRootForResidents(t)
+	defer parent.Stop()
+
+	dev := &outputTestDevice{name: "fakeout"}
+	if err := parent.io.RegisterDevice(dev); err != nil {
+		t.Fatal(err)
+	}
+
+	out := parent.executeOutputSendTool(agentAPI.ToolCall{
+		Name:      "output_send__fakeout",
+		Arguments: map[string]interface{}{"payload": "只给 payload，不给 type"},
+	})
+	if out != "ok" {
+		t.Fatalf("省略 type 时应默认 text 并发送成功，得到 %q", out)
+	}
+	if len(dev.sent) != 1 {
+		t.Fatalf("通道应收到 1 次输出，得到 %d", len(dev.sent))
+	}
+	if args, _ := dev.sent[0]["args"].(map[string]interface{}); args["type"] != "text" {
+		t.Fatalf("缺省类型应为 text，实际 %v", args["type"])
+	}
+
+	// 工具 schema：required 只应含 payload
+	var found bool
+	for _, td := range parent.buildToolDefs() {
+		entry, _ := td.(map[string]interface{})
+		fn, _ := entry["function"].(map[string]interface{})
+		if n, _ := fn["name"].(string); n != "output_send__fakeout" {
+			continue
+		}
+		found = true
+		params, _ := fn["parameters"].(map[string]interface{})
+		req, _ := params["required"].([]string)
+		if len(req) != 1 || req[0] != "payload" {
+			t.Fatalf("output_send 的 required 应只有 payload，实际 %v", req)
+		}
+	}
+	if !found {
+		t.Fatal("未生成 output_send__fakeout 工具")
+	}
+}
+
+// 空 payload 仍应被拦（这条判据是对的：不知道发什么不能放过）。
+func TestOutputSendStillRequiresPayload(t *testing.T) {
+	parent, _, _ := newRootForResidents(t)
+	defer parent.Stop()
+
+	if err := parent.io.RegisterDevice(&outputTestDevice{name: "fakeout"}); err != nil {
+		t.Fatal(err)
+	}
+	out := parent.executeOutputSendTool(agentAPI.ToolCall{
+		Name:      "output_send__fakeout",
+		Arguments: map[string]interface{}{"type": "text"},
+	})
+	if !strings.Contains(out, "payload 不能为空") {
+		t.Fatalf("空 payload 应被拦，得到 %q", out)
+	}
+}

internal/agent/core/resident.go

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

internal/agent/core/resident_test.go

@@ -142,6 +142,11 @@ func TestResident_LifecycleAndNoOrphans(t *testing.T) {
 	if ch, _ := reg.Lookup("sub/in"); ch.Owner != "" {
 		t.Fatalf("销毁后 inputch 应回到未分配：%+v", ch)
 	}
+	// 父的入站 inputch（child/<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 {
 		t.Fatal("重复销毁应报错")
 	}
@@ -159,6 +164,9 @@ func TestResident_LifecycleAndNoOrphans(t *testing.T) {
 		if _, err := osStat(filepath.Join(dir, "residents", id)); err == nil {
 			t.Fatalf("子 %s 的 temp 目录应被丢弃", id)
 		}
+		if _, ok := reg.Lookup(inboundChannelName(id)); ok {
+			t.Fatalf("父退出后子 %s 的入站 inputch 应被注销", id)
+		}
 	}
 }
 

internal/agent/core/scheduler.go

@@ -229,6 +229,10 @@ type SchedulerStats struct {
 	Executed uint64
 	// Rejected 是因队列满（或深度超限）而未被接纳的次数。
 	Rejected uint64
+	// Backpressure 是就绪队列满、输入被挡回 channel 的次数
+	// （设计 §4.4 / §11.4 Q4：满时阻塞发送方，**必须计数并打日志**）。
+	// 与 Rejected 的区别：Rejected 是「丢了」，Backpressure 是「暂时不收、发送方在等」。
+	Backpressure uint64
 	// Suspended / Resumed 是挂起与恢复的次数。
 	// 不变量：系统排空后 Suspended == Resumed（挂起必然被恢复），
 	// 因此两者各自只在**一处**计数（suspend / resumeTask）。
@@ -284,7 +288,13 @@ func (a *Agent) schedulerStatus() sdk.SchedulerStatus {
 		Rejected:          snap.Stats.Rejected,
 		Suspended:         snap.Stats.Suspended,
 		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 {
 		out.Running = &sdk.SchedulerTask{
@@ -320,6 +330,9 @@ type scheduler struct {
 	// critical 报告运行任务是否在不可抢占临界区（如记忆整理）。
 	// 由于 interceptLoop 要读它，必须是原子的：帧仍只由调度器读写。
 	critical atomic.Bool
+	// backpressured 记录「就绪队列满」这一状态的翻转，用于只打一次日志。
+	// 满着的时候 pumpInbox 每轮都会走到，逐轮打日志会把日志刷爆。
+	backpressured bool
 	// wake 用于把空闲的调度器叫醒：pendingInterrupts 不是 channel，
 	// 没有这个信号时“空闲时到达的中断”会一直等下一次输入（设计 §5.1 ③）。
 	wake chan struct{}
@@ -381,6 +394,28 @@ func (s *scheduler) hasRoom() bool {
 	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 与入队时刻（调用方持锁）。
 func (s *scheduler) allocateIDLocked(t *Task) {
 	s.seq++
@@ -490,10 +525,22 @@ func (s *scheduler) interruptCountLocked() int {
 // setImmediateLocked 登记一个应“立即运行”的抢占者。
 //
 // 槽只有一格：若已有抢占者且新的级别更高，旧的降级入队；否则新的入队。
-func (s *scheduler) setImmediateLocked(t *Task) {
+// 返回 true 表示 t **确实占住了 immediate 槽**；false 表示它被降级进了自己的
+// 级别队列（immediate 是单槽，这是设计要求的降级分支，见设计 §2「至多一个」）。
+//
+// 调用方必须用返回值决定是否计入 PreemptsByLevel：那条计数器的语义是
+// 「进入 immediate 的次数」，被降级的中断从未进过 immediate。
+// setImmediateLocked 尝试把 t 放进 immediate 槽。
+//
+// 返回 true：t 已占住 immediate（若原有抢占者被顶掉，它**已被**降级入队）。
+// 返回 false：t 没有进 immediate，且本函数**未动 t** —— 调用方负责按级别入队。
+//
+// 把「降级入队」的责任留给调用方，是为了让「到底入队了几次」只有一个出口：
+// 早先由本函数在返回 false 前自行入队，调用方又照着 false 再入一次，
+// 同一任务就会在队列里出现两份（实测：中断任务被执行两次、Executed 虚高）。
+func (s *scheduler) setImmediateLocked(t *Task) bool {
 	if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) {
-		s.enqueueInterruptLocked(t)
-		return
+		return false
 	}
 	if s.immediate != nil {
 		s.enqueueInterruptLocked(s.immediate)
@@ -501,6 +548,7 @@ func (s *scheduler) setImmediateLocked(t *Task) {
 	s.allocateIDLocked(t)
 	s.stats.Enqueued++
 	s.immediate = t
+	return true
 }
 
 func removeTask(list []*Task, target *Task) []*Task {
@@ -571,11 +619,16 @@ func (s *scheduler) registerInterrupt(t *Task) bool {
 	arm := false
 	if !critical && canPreempt(t, running) {
 		if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown {
-			arm = true
-			s.preemptArmed = true
-			s.preemptLevel = t.Level
-			s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
-			s.setImmediateLocked(t)
+			// 只有**真的占住 immediate 槽**才算一次抢占，才计入 PreemptsByLevel：
+			// immediate 是单槽，若它被另一个更高级的抢占者占着，t 会走上而下的
+			// 「否则入队」分支——那种情况 t 从未进入 immediate（否则同一安全点前
+			// 到达两条同级中断时该计数会高估）。
+			if s.setImmediateLocked(t) {
+				arm = true
+				s.preemptArmed = true
+				s.preemptLevel = t.Level
+				s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
+			}
 		}
 	}
 	if !arm {
@@ -606,6 +659,54 @@ func (s *scheduler) clearPreempt() {
 	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 保存现场。
 //
 // 深度上界是**结构推论**（= 中断级数），不是配置项：安全点上的 canSuspend 已提前
@@ -638,6 +739,37 @@ func (s *scheduler) canSuspend() bool {
 	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 标记任务执行结束。
 func (s *scheduler) done(t *Task) {
 	s.mu.Lock()
@@ -796,9 +928,11 @@ func (a *Agent) schedulerLoop() {
 			// 无待办：阻塞等新输入、新中断（wake）或退出。
 			select {
 			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:
-				a.sched.enqueue(newSelfTask(msg))
+				_ = a.sched.enqueue(newSelfTask(msg))
 			case <-a.sched.wake:
 				// 中断已入 pendingInterrupts，回到循环顶部重新挑选。
 			case <-a.ctx.Done():
@@ -825,15 +959,28 @@ func (a *Agent) pumpInbox() {
 	for a.sched.hasRoom() {
 		select {
 		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:
-			a.sched.enqueue(newSelfTask(msg))
+			// 自循环输入没有同步调用方，满时记一次背压即可。
+			if !a.sched.enqueue(newSelfTask(msg)) {
+				a.sched.noteBackpressure()
+			}
 		case <-a.ctx.Done():
 			return
 		default:
+			a.sched.clearBackpressure()
 			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。

internal/agent/core/scheduler_rearm_test.go

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

internal/agent/core/task.go

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

internal/agent/core/tooldefs.go

@@ -102,7 +102,14 @@ func (a *Agent) buildSystemPrompt(memContext string, userInput string) string {
 	prompt += "- 同步通道（webui / cli / 终端）：直接返回纯文本，内核会把文本交给等待方显示，无需调用工具。\n"
 	prompt += "- 异步通道（qq / wechat / 群聊等）：返回纯文本**【不会】**自动送达用户，必须调用 output_send__{通道名} 工具（注意 meta 里带上正确的 user_id 或 group_id）才能真正把消息发出去。\n"
 	prompt += "- 不确定当前通道的发送方式时，先用 output_send__{通道名}_help 查看该通道的 meta 格式和 type 枚举，再决定。\n"
-	prompt += "- 每轮对话**通常只需调用一次** output_send__{通道名} 即可完成回复。仅在内容确实超过单条消息长度上限（如 >4000 字）时才拆分为多条；拆分时每条应是完整段落，不要碎片化。\n"
+	// ❗这里**不得**限制"一轮只能发一次"。设计上输出是 agent 的**主动调用**：
+	// 收到一次输入后，可以往**任意（已授权的）通道**发**任意多次**（分段播报、
+	// 先回执后结论、同时通知多个通道都合法）。此前这里写着"每轮对话通常只需调用
+	// 一次 output_send"——那是一条**凭空的限制**，会让模型自己收起合理的多次输出。
+	// 真正需要提醒的只有两件事：单条长度上限（超长拆成完整段落）与"别反复重发
+	// 完全相同的内容"（自律，不是判据）。
+	prompt += "- **输出次数与目标通道由你自己决定**：一次输入可以对同一通道发多条（先回执后结论、分步播报、分段长文），也可以同时发到多个通道（例如同时通知 webui 与 qq）。**没有任何「一轮只能发一次」的限制。**\n"
+	prompt += "- 输出时只需注意两点：单条消息的长度上限（超长就拆成完整段落，不要碎片化）；别反复重发**完全相同**的内容（那是浪费，不是限制）。\n"
 	prompt += "- 需要多步执行的长任务：**必须先**向当前对话通道发一条确认消息告诉用户已收到（异步通道用输出门工具，同步通道直接返回文本），**然后再**执行具体排查工具。确认消息不代表任务完成，发出后仍需继续执行实际工具并最终汇报结果。\n"
 	prompt += "- 用户从其他渠道发来「在哪里/怎么样了」这类追问时，先回忆上次任务的通道与上下文，再回同一通道。"
 
@@ -624,7 +631,7 @@ func (a *Agent) buildToolDefs() []interface{} {
 			"type": "function",
 			"function": map[string]interface{}{
 				"name":        "output_send__" + ch.Name,
-				"description": desc + "。能力: " + capStr + "。payload 为消息载荷，meta 为 JSON 发送元数据，type 为载荷类型。用 _help 查看 meta 格式和 type 枚举。",
+				"description": desc + "。能力: " + capStr + "。payload 为消息载荷（type 默认 text，可省略），meta 为 JSON 发送元数据。用 _help 查看 meta 格式与 type 枚举。",
 				"parameters": map[string]interface{}{
 					"type": "object",
 					"properties": map[string]interface{}{
@@ -638,10 +645,10 @@ func (a *Agent) buildToolDefs() []interface{} {
 						},
 						"type": map[string]interface{}{
 							"type":        "string",
-							"description": "载荷类型，用 channel._help 查看支持的枚举值",
+							"description": "载荷类型，默认 text；其它枚举用 channel._help 查看",
 						},
 					},
-					"required": []string{"payload", "type"},
+					"required": []string{"payload"},
 				},
 			},
 		})

internal/devicebridge/client/bridge.go

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

internal/lua/sdk/sdk.lua

@@ -69,6 +69,65 @@ 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)
@@ -101,6 +160,9 @@ function sdk.doc.query(text, top_k) return {} end
 -- 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
@@ -174,6 +236,24 @@ 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 = {}
 

internal/meta/meta.go

@@ -28,6 +28,7 @@ var (
 	//
 	// ❗main 上此值始终是**下一个未发布中版本**，不随 patch 发布变动
 	//（见 docs/git-branching.md §2.1）；已发布的版本号看对应的 release/vX.Y.x 与 tag。
+	// 1.3.10：去掉提示词里"每轮只能发一次 output_send"的凭空限制；type 缺省即 text。
 	// 1.3.9：驻留子的「轮次」不再是恒 0（info() 此前没填 Rounds）。
 	// 1.3.8：inputch 划给子后输入只流向子（补上"进内核之前"的输入路由）。
 	// 1.3.7：驻留子继承父的输出通道（此前子侧 childIO 空壳 ⇒ 子不会发消息）。
@@ -36,7 +37,15 @@ var (
 	// 1.3.5：系统提示词（人格卡）支持版本占位符 —— 人格卡是配置项，写死版本号
 	//        会随发版说谎（线上写 v1.0.3、内核 1.3.x，agent 就自报 1.0.3）。
 	//        支持 {{kernel_version}} / {{kernel_commit}} / {{sdk_version}}。
-	Version = "1.3.9"
+	// 1.3.11：Lua 插件桥全量对齐 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 = "unknown"

internal/plugin/lua_plugin.go

@@ -1,6 +1,7 @@
 package plugin
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -9,9 +10,10 @@ import (
 	"strings"
 	"sync"
 
-	lua "github.com/yuin/gopher-lua"
+	agentEvents "gitcode.com/JianFeeeee/HomeAgent/internal/events"
 	luaSDK "gitcode.com/JianFeeeee/HomeAgent/internal/lua/sdk"
 	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
+	lua "github.com/yuin/gopher-lua"
 )
 
 type toolReg struct {
@@ -40,7 +42,16 @@ type luaPlugin struct {
 	stages    map[sdk.Stage]*stageReg
 	outputChs map[string]*outputChReg
 	inputDefs map[string]sdk.ChannelDef
-	mu        sync.Mutex
+	// subs 是本插件注册的事件订阅取消函数；Stop 时兜底取消，
+	// 避免 L 已 Close 后残留回调被触发（use-after-close）。
+	// 用独立的 subsMu 而非 mu：subscribe 会在 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) {
@@ -287,6 +298,19 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
 		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 {
 		s.InjectText(L.CheckString(1), L.CheckString(2), L.CheckString(3))
 		return 0
@@ -300,6 +324,67 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
 		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 外部插件面完全对齐）----
 	// 约定：结果型返回 (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 {
 			tbl.ForEach(func(_, v lua.LValue) {
 				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{
-						Subject:     t2.RawGetString("subject").String(),
-						Relation:    t2.RawGetString("relation").String(),
-						Object:      t2.RawGetString("object").String(),
-						Confidence:  float64(lua.LVAsNumber(t2.RawGetString("confidence"))),
-						SubjectType: t2.RawGetString("subject_type").String(),
-						ObjectType:  t2.RawGetString("object_type").String(),
+						Subject:      t2.RawGetString("subject").String(),
+						Relation:     t2.RawGetString("relation").String(),
+						Object:       t2.RawGetString("object").String(),
+						Confidence:   float64(lua.LVAsNumber(t2.RawGetString("confidence"))),
+						SubjectType:  t2.RawGetString("subject_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 {
 		if dm := s.DocMemory(); dm != nil {
-			tbl := L.CheckTable(1)
-			if err := dm.Insert(&sdk.Doc{
-				ID:      tbl.RawGetString("id").String(),
-				Title:   tbl.RawGetString("title").String(),
-				Content: tbl.RawGetString("content").String(),
-			}); err != nil {
+			if err := dm.Insert(docFromLua(L.CheckTable(1))); err != nil {
+				return pushErr(err)
+			}
+		}
+		return pushNil()
+	}))
+	// insert_with_media（1.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)
 			}
 		}
@@ -503,10 +601,11 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
 		if tmem := s.TextMemory(); tmem != nil {
 			tbl := L.CheckTable(1)
 			if err := tmem.Append(sdk.TextEvent{
-				Role:      tbl.RawGetString("role").String(),
-				Content:   tbl.RawGetString("content").String(),
-				Timestamp: int64(lua.LVAsNumber(tbl.RawGetString("timestamp"))),
-				Channel:   tbl.RawGetString("channel").String(),
+				Role:        tbl.RawGetString("role").String(),
+				Content:     tbl.RawGetString("content").String(),
+				Timestamp:   int64(lua.LVAsNumber(tbl.RawGetString("timestamp"))),
+				Channel:     tbl.RawGetString("channel").String(),
+				Attachments: luaToAttachments(L, tbl.RawGetString("attachments")),
 			}); err != nil {
 				return pushErr(err)
 			}
@@ -700,6 +799,76 @@ func replaceSDKReal(L *lua.LState, t *lua.LTable, plg *luaPlugin, s *sdk.PluginS
 		}
 		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 {
@@ -724,13 +893,29 @@ func makeStageHandler(plg *luaPlugin, stage sdk.Stage, fn *lua.LFunction) sdk.St
 		defer plg.mu.Unlock()
 		L := plg.L
 		ctx := map[string]interface{}{
-			"raw_message": sc.RawMessage,
-			"user_id":     sc.UserID,
-			"group_id":    sc.GroupID,
-			"phase":       string(sc.Phase),
-			"llm_text":    sc.LLMText,
-			"final_text":  sc.FinalText,
-			"no_memory":   sc.NoMemory,
+			"raw_message":       sc.RawMessage,
+			"user_id":           sc.UserID,
+			"group_id":          sc.GroupID,
+			"phase":             string(sc.Phase),
+			"llm_text":          sc.LLMText,
+			"reasoning_content": sc.ReasoningContent,
+			"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 {
 			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 {
 		goDef.NoMemory = lua.LVAsBool(v)
 	}
+	goDef.ContextPolicy = defTbl.RawGetString("context_policy").String()
 	if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction {
 		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 {
 		chDef.NoMemory = lua.LVAsBool(v)
 	}
+	chDef.ContextPolicy = defTbl.RawGetString("context_policy").String()
 	if v := defTbl.RawGetString("cleaner"); v != nil && v.Type() == lua.LTFunction {
 		chDef.Cleaner = makeLuaCleaner(plg, v.(*lua.LFunction))
 	}
 	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{} 树。
 func jsonToIface(v interface{}) interface{} {
 	b, err := json.Marshal(v)
@@ -959,8 +1237,21 @@ func (p *luaPlugin) Start(s *sdk.PluginSDK) 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()
 	defer p.mu.Unlock()
+	p.closed = true
 
 	if p.tbl != nil {
 		fn := p.tbl.RawGetString("stop")

internal/plugin/lua_plugin_test.go

@@ -5,9 +5,10 @@ import (
 	"path/filepath"
 	"testing"
 
-	lua "github.com/yuin/gopher-lua"
 	internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config"
+	"gitcode.com/JianFeeeee/HomeAgent/internal/events"
 	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
+	lua "github.com/yuin/gopher-lua"
 )
 
 func TestTryLoadLua_Basic(t *testing.T) {
@@ -597,3 +598,66 @@ return plugin
 		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"})
+}

internal/plugin/lua_surface_test.go

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

internal/plugin/plugin.go

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

internal/plugins/remotedevice/ping_test.go

@@ -0,0 +1,53 @@
+package remotedevice
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+// 心跳回包必须**真的发出去**：pong 只有两个字节，且设备空闲时没有任何别的写
+// 会顺带把 bufio 缓冲刷出去——`writePong` 一旦忘了 Flush，pong 就永远留在
+// 服务端缓冲里。
+//
+// 这就是「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（服务端要先把设备登记进 conns，pong 才写得回来）。
+	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 内没收到 pong（writePong 忘了 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)
+	}
+}

internal/plugins/remotedevice/registry.go

@@ -606,7 +606,13 @@ func writeFrame(w *bufio.Writer, opcode byte, payload []byte) 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 {

internal/sdk/status.go

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

third_party/homeagent-sdk/sdk/lua/sdk.lua

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