jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-09-21 17:38:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: LLM 工具循环 400、中断消息注入、ConPTY 终端支持

Browse Source 
- agent: 工具轮请求尾部补 user 占位(zen 网关强制),tool 消息正确配对
- agent: 工具提醒/中断以 system 角色注入并带 [中断消息] 前缀,不进用户履历;系统提示词说明中断消息格式
- agentcli: 基于 ConPTY 的交互式终端(ptywin fork),terminal_create/read/write/resize/close/watch
- webui: server 输出通道适配器(保留 reasoning_content/disable_thinking)
- GUI: 沉浸式标题栏、icon 圆角重制、mascot 等打磨
This commit is contained in:
JianFeeeee
2026-08-14 00:48:40 +08:00
parent 816597caac
commit 147d0baaf9
43 changed files with 4670 additions and 1478 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
internal/agent/core/agent.go
View File

@ -100,6 +100,9 @@ type Agent struct {
// 当前轮次的非文本媒体数据（图片/音频），供 describe_image 等工具访问
pendingMedia map[string]interface{}
// 当前输入是否为工具提醒/中断（以 system 角色注入，避免被当成用户消息）
interruptInput bool
// 非文本输入处理配置
inputCfg types.InputProcessingConfig

77
internal/agent/core/agent_helpers_test.go
View File

@ -86,6 +86,83 @@ func TestDocToTriplesEmptyContent(t *testing.T) {
}
}
// Phase 2: 归档上下文文档不得产出模板垃圾（context_archived 来源/主题模板三元组）
func TestDocToTriplesArchivedContext(t *testing.T) {
doc := &document.Doc{
Summary: "来自 2 个来源的 5 条对话 (qq, webui) 涉及: 天气, 测试",
Content: "[15:04] qq: 今天天气怎么样\n[15:05] agent: 今天天气很好",
Source: "context_archived",
Meta: map[string]string{"is_archived_context": "true"},
}
triples := docToTriples(doc, nil)
for _, tr := range triples {
if tr.Subject == "文档" && tr.Relation == "来源" && tr.Object == "context_archived" {
t.Errorf("archived context must not write 来源 triple: %+v", tr)
}
if tr.Subject == "文档" && tr.Relation == "主题" {
t.Errorf("archived context must not write 主题 template triple: %+v", tr)
}
}
}
// Phase 2: 模板化摘要（summarizeEntries 生成）不得作为主题写入
func TestDocToTriplesTemplateSummary(t *testing.T) {
doc := &document.Doc{
Summary: "来自 3 个来源的 10 条对话 (a, b, c) 涉及: 关键词1, 关键词2, 关键词3",
Content: "[10:00] a: 你好",
Source: "manual",
}
triples := docToTriples(doc, nil)
for _, tr := range triples {
if tr.Subject == "文档" && tr.Relation == "主题" {
t.Errorf("template summary must not be written as 主题 triple: %+v", tr)
}
}
// 但非归档来源仍保留 来源 三元组
foundSource := false
for _, tr := range triples {
if tr.Subject == "文档" && tr.Relation == "来源" && tr.Object == "manual" {
foundSource = true
}
}
if !foundSource {
t.Errorf("non-archived source should still produce 来源 triple")
}
}
// Phase 2: 过长摘要不得写入主题
func TestDocToTriplesLongSummary(t *testing.T) {
long := ""
for i := 0; i < 100; i++ {
long += "很长的摘要内容片段重复拼接"
}
doc := &document.Doc{
Summary: long,
Content: "[10:00] a: 你好",
Source: "test",
}
triples := docToTriples(doc, nil)
for _, tr := range triples {
if tr.Subject == "文档" && tr.Relation == "主题" {
t.Errorf("overlong summary must not be written as 主题 triple")
}
}
}
func TestIsTemplateSummary(t *testing.T) {
if !isTemplateSummary("来自 2 个来源的 5 条对话 (qq, webui) 涉及: 天气") {
t.Errorf("template summary not recognized")
}
if isTemplateSummary("今天天气很好") {
t.Errorf("plain summary wrongly recognized as template")
}
if !isTemplateSummary("") {
t.Errorf("empty summary should be treated as template")
}
}
func TestTruncateStr(t *testing.T) {
tests := []struct {
input string

36
internal/agent/core/distill.go
View File

@ -4,6 +4,7 @@ import (
"fmt"
"log"
"runtime/debug"
"strings"
"time"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
@ -397,15 +398,19 @@ func docToTriples(doc *document.Doc, embedder nlp.Vectorizer) []memory.Triple {
return nil
}
// 文档元数据
triples = append(triples, memory.Triple{
Subject: "文档",
SubjectType: "Concept",
Relation: "主题",
Object: doc.Summary,
ObjectType: "Topic",
Confidence: 1.0,
})
isArchivedContext := doc.Meta != nil && doc.Meta["is_archived_context"] == "true"
// 文档元数据:仅当 summary 合理(非空、非模板化、长度适中)时才写「主题」
if !isArchivedContext && doc.Summary != "" && len([]rune(doc.Summary)) < 80 && !isTemplateSummary(doc.Summary) {
triples = append(triples, memory.Triple{
Subject: "文档",
SubjectType: "Concept",
Relation: "主题",
Object: doc.Summary,
ObjectType: "Topic",
Confidence: 1.0,
})
}
// NLP 通用提取
e := nlp.NewExtractor(nil)
@ -422,7 +427,8 @@ func docToTriples(doc *document.Doc, embedder nlp.Vectorizer) []memory.Triple {
}
}
if doc.Source != "" {
// 仅当来源非归档上下文且非空时写「来源」——归档文档写死模板三元组属于垃圾
if doc.Source != "" && doc.Source != "context_archived" {
triples = append(triples, memory.Triple{
Subject: "文档",
SubjectType: "Concept",
@ -436,6 +442,16 @@ func docToTriples(doc *document.Doc, embedder nlp.Vectorizer) []memory.Triple {
return triples
}
// isTemplateSummary 识别 summarizeEntries 生成的模板化摘要
// （形如「来自 N 个来源的 M 条对话 (src1, src2) 涉及: kw1, kw2」），
// 这类摘要无独立信息量,不应作为「主题」实体写入图库。
func isTemplateSummary(s string) bool {
if s == "" {
return true
}
return strings.HasPrefix(s, "来自 ") && strings.Contains(s, "条对话")
}
func (a *Agent) emitMemoryCandidate(source, input, response string, toolResults []ToolResultItem, toolsUsed []string) {
a.io.EmitOutput("memory", "memory_candidate", map[string]interface{}{
"source": source,

31
internal/agent/core/eventloop.go
View File

@ -287,6 +287,16 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) {
}
}
// 工具提醒/中断（terminal_watch、timer 等）不是用户发言：
// 以 system 角色注入 LLM，且不写入用户对话履历。
isInterrupt, _ := evt.Payload["interrupt"].(bool)
a.mu.Lock()
a.interruptInput = isInterrupt
a.mu.Unlock()
if isInterrupt {
noMemory = true
}
stageCtx := a.stageCtxFromInput(input, evt.Source, "")
stageCtx.Extra["input_source"] = evt.Source
stageCtx.Extra["output_channel"] = evt.OutputChannel
@ -320,11 +330,13 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) {
log.Printf("[agent] pruned %d low-relevance events to document memory", archived)
}
a.context.Append(ContextEvent{
Timestamp: start,
Source: evt.Source,
Input: input,
})
if !isInterrupt {
a.context.Append(ContextEvent{
Timestamp: start,
Source: evt.Source,
Input: input,
})
}
response, toolsUsed, toolResults, err := a.process(input, stageCtx)
if err != nil {
@ -380,7 +392,6 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
if stageCtx.TokenUsage != nil {
payload["usage"] = stageCtx.TokenUsage
}
if evt.ResponseCh != nil {
evt.ResponseCh <- &agentIO.OutputEvent{
RequestID: evt.RequestID,
@ -392,11 +403,15 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
}
}
a.publishEvent(events.EventAgentOutput, map[string]interface{}{
out := map[string]interface{}{
"content": response,
"channel": ch,
"source": evt.Source,
})
}
if stageCtx.ReasoningContent != "" {
out["reasoning_content"] = stageCtx.ReasoningContent
}
a.publishEvent(events.EventAgentOutput, out)
stageCtx.Phase = sdk.StageAfterOutput
a.runStage(sdk.StageAfterOutput, stageCtx)
}

71
internal/agent/core/process.go
View File

@ -27,6 +27,14 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
tools := a.buildToolDefs()
msgs := a.buildMessages(sysPrompt, input, budget.ContextTokens)
// 工具提醒（interrupt）：以 system 角色注入，不让模型误认为用户发言
if a.interruptInput {
last := msgs[len(msgs)-1]
last.Role = "system"
last.Content = "[中断消息] " + last.Content
msgs[len(msgs)-1] = last
a.interruptInput = false
}
if blocks, ok := stageCtx.Extra["media_blocks"].([]agentAPI.ContentBlock); ok && len(blocks) > 0 {
if len(msgs) > 0 {
msgs[len(msgs)-1].Blocks = blocks
@ -56,7 +64,16 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
for _, interrupt := range a.drainInterrupts() {
msgs = append(msgs, agentAPI.Message{
Role: "system",
Content: interrupt,
Content: "[中断消息] " + interrupt,
})
}
// zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验),
// 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。
if last := msgs[len(msgs)-1]; last.Role != "user" {
msgs = append(msgs, agentAPI.Message{
Role: "user",
Content: "请根据以上工具结果继续。",
})
}
@ -177,6 +194,13 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
}
a.publishEvent(events.EventAgentLLMChain, chainPayload)
if resp.ReasoningContent != "" {
a.publishEvent(events.EventReasoning, map[string]interface{}{
"content": resp.ReasoningContent,
"channel": a.currentOutputChannel,
})
}
if len(resp.ToolCalls) == 0 {
return resp.Content, toolsUsed, toolResults, nil
}
@ -185,15 +209,16 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
for _, tc := range resp.ToolCalls {
if len(a.interceptCh) > 0 {
for _, interrupt := range a.drainInterrupts() {
msgs = append(msgs, agentAPI.Message{Role: "system", Content: interrupt})
msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
}
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": a.resolveToolPlugin(tc.Name),
"args": tc.Arguments,
"status": "interrupted",
"reason": "user interrupt before execution",
})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": a.resolveToolPlugin(tc.Name),
"args": tc.Arguments,
"status": "interrupted",
"reason": "user interrupt before execution",
"channel": a.currentOutputChannel,
})
break
}
@ -208,13 +233,14 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name)
msgs = append(msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}})
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": pluginName,
"args": tc.Arguments,
"result": result,
"status": "denied",
})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": pluginName,
"args": tc.Arguments,
"result": result,
"status": "denied",
"channel": a.currentOutputChannel,
})
continue
}
tc.Arguments = stageCtx.ToolCalls[0].Arguments
@ -248,16 +274,17 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
a.publishEvent(events.EventToolCall, map[string]interface{}{
"tool": tc.Name,
"plugin": pluginName,
"args": tc.Arguments,
"result": result,
"status": "ok",
"tool": tc.Name,
"plugin": pluginName,
"args": tc.Arguments,
"result": result,
"status": "ok",
"channel": a.currentOutputChannel,
})
if len(a.interceptCh) > 0 {
for _, interrupt := range a.drainInterrupts() {
msgs = append(msgs, agentAPI.Message{Role: "system", Content: interrupt})
msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
}
break
}

8
internal/agent/core/stage.go
View File

@ -12,6 +12,14 @@ import (
)
func (a *Agent) runStage(stage sdk.Stage, ctx *sdk.StageContext) bool {
payload := map[string]interface{}{
"phase": string(stage),
"channel": a.currentOutputChannel,
}
if ctx != nil && len(ctx.ToolCalls) > 0 {
payload["tool"] = ctx.ToolCalls[0].Name
}
a.publishEvent(events.EventStage, payload)
if a.stageHost == nil {
return false
}

9
internal/agent/core/tooldefs.go
View File

@ -49,12 +49,13 @@ func (a *Agent) buildSystemPrompt(memContext string, userInput string) string {
}
}
prompt += "\n\n【输出规则】你有多组输出门工具（type=output），每个对应一个输出通道。回复用户时必须调用对应的 output_send__{通道名} 工具。\n"
prompt += "- payload 参数是消息载荷（文本直接填文字），type 指定载荷类型（text/voice/image/file），meta 是 JSON 发送元数据（群号/用户号等）。\n"
prompt += "\n\n【中断消息】长任务执行期间，工具/插件/定时器等会通过中断机制向你发送提醒（如 QQ 新消息、终端输出到达、定时器到点等）。中断消息以 system 角色注入，内容带 [中断消息] 前缀，**不是用户发言，但也必须认真处理**：优先停下当前长任务，针对中断内容作出响应或决定继续执行。不要忽略带 [中断消息] 前缀的 system 消息。"
prompt += "\n\n【输出规则】回复会自动发送到用户的输入来源通道，直接返回纯文本即可送达，无需调用任何工具。\n"
prompt += "- 输出门工具 output_send__{通道名} 用于主动向指定通道推送消息（如群发、主动通知、向其他通道发言），不是回复的必要步骤。除非用户要求在别的通道发送，否则不要使用。\n"
prompt += "- 用 output_send__{通道名}_help 查看该通道的 meta 格式和 type 枚举。\n"
prompt += "- 同一轮对话中可多次调用输出门工具。长消息应当分多次发出，而不是一口气发完。\n"
prompt += "- 直接返回纯文本不会到达任何用户端。\n"
prompt += "- 需要多步执行的长任务：**必须先**用 output_send__ 发一条确认消息告诉用户已收到（如「好的我去看看～」），**然后再**执行具体排查工具。确认消息不代表任务完成，发出后仍需继续执行实际工具并最终汇报结果。"
prompt += "- 需要多步执行的长任务：**必须先**用输出门工具向当前输入通道发一条确认消息告诉用户已收到（如「好的我去看看～」，也可以直接返回文本），**然后再**执行具体排查工具。确认消息不代表任务完成，发出后仍需继续执行实际工具并最终汇报结果。"
if a.indexer != nil {
prompt += "\n\n" + a.indexer.BuildToolPrompt()

8
internal/config/registry.go
View File

@ -568,9 +568,9 @@ func (r *ConfigRegistry) seedDBValues(dataDir string) {
WebUI 概览页展示你的立绘，可通过 /mascot.webp 直接访问。如输出通道支持图片引用，可借此发送自己的立绘。
回复默认发送到用户的输入来源，无需额外工具。
输出回复请使用 output_send__{通道名} 工具，content 为 JSON 字符串。用 output_list_channels 查看可用通道。
使用 output_send__{通道名}_help 查看每个通道的 JSON 格式说明。
回复会自动发送到用户的输入来源通道，直接返回纯文本即可送达，无需额外工具。
输出门工具 output_send__{通道名} 仅用于主动向指定通道推送消息（群发、主动通知、向其他通道发言），不是回复的必要步骤。用 output_list_channels 查看可用通道。
使用 output_send__{通道名}_help 查看每个通道的格式说明。
输出通道可多次调用，长消息应当分多次发出而不是一口气发完。
当用户上传图片或音频时，系统会自动附着媒体内容。如果模型不支持直接处理多媒体，请调用对应的媒体处理工具。`)
@ -646,7 +646,7 @@ func (r *ConfigRegistry) seedCoreDefs(dataDir string) {
reg(ConfigDef{Key: "core.agent.review_interval", Default: "120m", Type: "duration", DisplayName: "关系复审间隔", Description: "三元组关系复审的执行间隔", Category: "agent"})
reg(ConfigDef{Key: "core.agent.merge_interval", Default: "120m", Type: "duration", DisplayName: "实体合并检测间隔", Description: "实体合并检测（LLM 裁决）的执行间隔", Category: "agent"})
reg(ConfigDef{Key: "core.agent.workdir", Default: "", Type: "string", DisplayName: "工作目录", Description: "Agent 命令执行的默认工作目录（如 cmd_run 工具的 fallback），留空使用内核所在目录", Category: "agent"})
reg(ConfigDef{Key: "core.agent.embedding_model_path", Default: "", Type: "string", DisplayName: "预训练词嵌入模型路径", Description: "预训练词嵌入模型路径（word2vec 文本格式），支持逗号分隔多个模型。空则使用 TF-IDF 回退。修改后需重启生效。", Category: "agent"})
reg(ConfigDef{Key: "core.agent.embedding_model_path", Default: "", Type: "string", DisplayName: "预训练词嵌入模型路径", Description: "预训练词嵌入模型路径（word2vec 文本格式），支持逗号分隔多个模型。路径后可加 #topN 规格只加载前 N 个词向量（如 /data/cc.zh.300.vec#top50000）以控制常驻内存，词频降序命中覆盖绝大部分文本。空则使用 TF-IDF 回退。修改后需重启生效。", Category: "agent"})
reg(ConfigDef{Key: "core.agent.onnx_model_path", Default: "", Type: "string", DisplayName: "ONNX 模型路径", Description: "依存句法分析 ONNX 模型文件路径。留空使用二进制内嵌模型/规则引擎。修改后需重启生效。", Category: "agent"})
reg(ConfigDef{Key: "core.agent.system_prompt", Default: "", Type: "text", DisplayName: "系统身份提示词", Description: "Agent 的系统提示词，定义身份和行为规则。留空则使用编译时内置默认值。修改后需重启生效。", Category: "agent"})

1
internal/events/bus.go
View File

@ -16,6 +16,7 @@ const (
EventReasoning EventType = "reasoning"
EventStage EventType = "stage"
EventSystem EventType = "system"
EventTerminalOutput EventType = "terminal_output"
EventAll EventType = "*"
)

103
internal/lua/adapters/server.lua Normal file
View File

@ -0,0 +1,103 @@
local adapter = {}
adapter.name = "server"
adapter.version = "1.0.0"
adapter.endpoint = "/chat/completions"
adapter.headers = {}
-- 专用于 zen 兼容网关（thinking 模式要求回传 reasoning_content）。
-- 关键：不删除 disable_thinking（homeagent 置 true 时网关关闭 thinking，
-- 从而不再强制要求 reasoning_content 回传）；同时保留已有 reasoning_content 双保险。
function adapter.transform_request(raw_body)
local ok, req = pcall(json.decode, raw_body)
if not ok then return raw_body end
req.extra_body = nil
return json.encode(req)
end
function adapter.transform_response(raw_body)
local ok, resp = pcall(json.decode, raw_body)
if not ok or resp == nil then return raw_body end
local unified = {
content = "",
finish_reason = "",
token_usage = { prompt = 0, completion = 0, total = 0 }
}
if type(resp.usage) == "table" then
unified.token_usage.prompt = resp.usage.prompt_tokens or 0
unified.token_usage.completion = resp.usage.completion_tokens or 0
unified.token_usage.total = resp.usage.total_tokens or 0
end
if type(resp.choices) == "table" and #resp.choices > 0 then
local ch = resp.choices[1]
if type(ch.message) == "table" then
unified.content = ch.message.content or ""
if ch.message.reasoning_content then
unified.reasoning_content = ch.message.reasoning_content
end
if type(ch.message.tool_calls) == "table" then
local tcs = {}
for _, tc in ipairs(ch.message.tool_calls) do
local fn = tc["function"]
local name = tc.name
local raw_args = tc.arguments
if type(fn) == "table" then
name = fn.name or name
raw_args = fn.arguments or raw_args
end
local args = {}
if type(raw_args) == "table" then
args = raw_args
elseif type(raw_args) == "string" and raw_args ~= "" then
local args_ok, decoded = pcall(json.decode, raw_args)
if args_ok and type(decoded) == "table" then
args = decoded
elseif args_ok then
args = { value = decoded }
else
args = { raw = raw_args }
end
end
if name ~= nil and name ~= "" then
table.insert(tcs, {
id = tc.id,
type = tc.type or "function",
name = name,
arguments = args
})
end
end
unified.tool_calls = tcs
end
end
unified.finish_reason = ch.finish_reason or ""
end
return json.encode(unified)
end
function adapter.transform_stream_chunk(raw_chunk)
local ok, chunk = pcall(json.decode, raw_chunk)
if not ok then return "" end
if not chunk.choices or #chunk.choices == 0 then return "" end
local delta = chunk.choices[1].delta or {}
local fr = chunk.choices[1].finish_reason
local unified = {
content = delta.content or "",
done = (fr ~= nil)
}
if delta.reasoning_content then
unified.reasoning_content = delta.reasoning_content
end
if delta.tool_calls then
unified.tool_calls = delta.tool_calls
end
return json.encode(unified)
end
return adapter

1
internal/lua/vm.go
View File

@ -558,6 +558,7 @@ func (v *VM) writeBundledAdapters() error {
known := []string{
"openai", "anthropic", "deepseek", "gemini",
"github", "groq", "mistral", "ollama", "kimicode",
"server",
}
for _, name := range known {
srcPath := "adapters/" + name + ".lua"

6
internal/memory/document/document.go
View File

@ -183,6 +183,10 @@ func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.V
docVec = vec.Vectorize(summary + " " + content)
} else {
docVec = s.veczer.Vectorize(summary + " " + content)
}
meta := map[string]string{"content_hash": contentHash}
if source == "context_archived" {
meta["is_archived_context"] = "true"
}
doc := &Doc{
ID: id,
@ -195,7 +199,7 @@ func (s *Store) ContextToDoc(source string, entries []ContextEntry, vec vector.V
LastAccess: time.Now(),
AccessCount: 1,
Source: source,
Meta: map[string]string{"content_hash": contentHash},
Meta: meta,
Vector: docVec,
}
s.docs[id] = doc

31
internal/memory/pipeline/pipeline.go
View File

@ -193,11 +193,15 @@ func (d *Distiller) distillLoop() {
func (d *Distiller) distillOnce() {
d.mu.Lock()
cutoff := time.Now().AddDate(0, 0, -d.cfg.RetentionDays)
batchSize := d.cfg.BatchSize
if batchSize <= 0 {
batchSize = 50
}
// 每 tick 取前 N 条未蒸馏记录（无 RetentionDays 门槛），蒸馏成功才标记/移除
var toDistill []RawRecord
var remaining []RawRecord
for _, r := range d.records {
if r.CreatedAt.Before(cutoff) && !r.Distilled {
if !r.Distilled && len(toDistill) < batchSize {
toDistill = append(toDistill, r)
} else {
remaining = append(remaining, r)
@ -210,22 +214,29 @@ func (d *Distiller) distillOnce() {
return
}
batchSize := d.cfg.BatchSize
if batchSize <= 0 {
batchSize = 50
}
distilled := 0
for i := 0; i < len(toDistill); i += batchSize {
end := i + batchSize
if end > len(toDistill) {
end = len(toDistill)
}
d.distillBatch(toDistill[i:end])
if d.distillBatch(toDistill[i:end]) {
distilled += end - i
} else {
// 蒸馏失败：记录写回待处理队列，下次 tick 重试
d.mu.Lock()
d.records = append(toDistill[i:end], d.records...)
d.mu.Unlock()
}
}
d.cleanupRawFiles()
log.Printf("[memory] distilled %d records", len(toDistill))
if distilled > 0 {
log.Printf("[memory] distilled %d records", distilled)
}
}
func (d *Distiller) distillBatch(batch []RawRecord) {
// distillBatch 蒸馏一批记录，全部成功返回 true，任一失败返回 false（调用方重试）
func (d *Distiller) distillBatch(batch []RawRecord) bool {
var userContent, assistantContent string
sessionIDs := make(map[string]bool)
for _, r := range batch {
@ -245,8 +256,10 @@ func (d *Distiller) distillBatch(batch []RawRecord) {
}
if _, _, err := d.db.Commit(triples, sessionID, 0); err != nil {
log.Printf("[memory] distill commit: %v", err)
return false
}
}
return true
}
func (d *Distiller) cleanupRawFiles() {

66
internal/memory/pipeline/pipeline_test.go
View File

@ -1,6 +1,7 @@
package pipeline
import (
"fmt"
"os"
"path/filepath"
"testing"
@ -117,6 +118,71 @@ func TestDistillOnce(t *testing.T) {
}
}
// Phase 4: 新记录无需等待 RetentionDays，下一 tick 立即蒸馏（文档所述 10min 频率）
func TestDistillOnceFreshRecords(t *testing.T) {
db, err := memory.NewGraphDB(filepath.Join(t.TempDir(), "test.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
dir := t.TempDir()
d := NewDistiller(db, dir, DistillerConfig{
Interval: 10 * time.Minute,
RetentionDays: 7,
BatchSize: 50,
})
d.Append("sess1", "user", "我的名字是李四")
d.Append("sess1", "assistant", "你好李四！")
if len(d.records) != 2 {
t.Fatalf("expected 2 fresh records, got %d", len(d.records))
}
d.distillOnce()
if len(d.records) != 0 {
t.Errorf("fresh records should be distilled on next tick (no retention gate), got %d remaining", len(d.records))
}
// 二次蒸馏不重复（已蒸馏记录已被移除）
d.distillOnce()
if len(d.records) != 0 {
t.Errorf("second distill should be no-op, got %d records", len(d.records))
}
}
// Phase 4: BatchSize 限制每 tick 处理前 N 条，未蒸馏记录留待下个 tick
func TestDistillOnceBatchLimit(t *testing.T) {
db, err := memory.NewGraphDB(filepath.Join(t.TempDir(), "test.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
dir := t.TempDir()
d := NewDistiller(db, dir, DistillerConfig{
Interval: 10 * time.Minute,
RetentionDays: 7,
BatchSize: 3,
})
for i := 0; i < 10; i++ {
d.Append("sess1", "user", fmt.Sprintf("第 %d 条消息内容", i))
}
d.distillOnce()
if len(d.records) != 7 {
t.Fatalf("expected 7 records remaining after batch 3, got %d", len(d.records))
}
// 后续 tick 继续消化，最终全部蒸馏
for i := 0; i < 5 && len(d.records) > 0; i++ {
d.distillOnce()
}
if len(d.records) != 0 {
t.Errorf("all records should be distilled after several ticks, got %d remaining", len(d.records))
}
}
func TestExtractKeyTriples(t *testing.T) {
tests := []struct {
user string

35
internal/memory/static_embedder.go
View File

@ -129,12 +129,13 @@ func ensureModelFile(modelPath string) {
if modelPath == "" {
return
}
if _, err := os.Stat(modelPath); err == nil {
path, _ := parseModelSpec(modelPath)
if _, err := os.Stat(path); err == nil {
return
}
url := modelDownloadURL(modelPath)
log.Printf("[static_embedder] model %s not found, downloading from fastText...", modelPath)
if dlErr := downloadFastTextModel(modelPath, url); dlErr != nil {
url := modelDownloadURL(path)
log.Printf("[static_embedder] model %s not found, downloading from fastText...", path)
if dlErr := downloadFastTextModel(path, url); dlErr != nil {
log.Printf("[static_embedder] download failed: %v, will use TF-IDF fallback", dlErr)
} else {
log.Printf("[static_embedder] download ok")
@ -183,7 +184,24 @@ func (e *StaticEmbedder) loadAll(paths []string) error {
return firstErr
}
func (e *StaticEmbedder) load(path string, primary bool) error {
// parseModelSpec 解析模型路径规格：`path#top50000` 表示只加载前 50000 个词向量（按文件顺序，fastText
// 词频降序，前 N 词覆盖绝大多数文本命中），用于降低常驻内存；无规格返回原路径与 0（全量加载）。
func parseModelSpec(p string) (path string, topN int) {
path = p
if i := strings.IndexByte(p, '#'); i >= 0 {
path = p[:i]
spec := p[i+1:]
if strings.HasPrefix(spec, "top") {
if n, err := strconv.Atoi(strings.TrimPrefix(spec, "top")); err == nil && n > 0 {
topN = n
}
}
}
return path, topN
}
func (e *StaticEmbedder) load(spec string, primary bool) error {
path, topN := parseModelSpec(spec)
f, err := os.Open(path)
if err != nil {
return fmt.Errorf("open: %w", err)
@ -217,7 +235,11 @@ func (e *StaticEmbedder) load(path string, primary bool) error {
vecSum = make([]float64, dim)
}
loaded := 0
for scanner.Scan() {
if topN > 0 && loaded >= topN {
break
}
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
@ -244,6 +266,7 @@ func (e *StaticEmbedder) load(path string, primary bool) error {
}
count++
}
loaded++
}
if primary {
@ -263,7 +286,7 @@ func (e *StaticEmbedder) load(path string, primary bool) error {
e.loaded = true
}
log.Printf("[static_embedder] loaded %d words, dim=%d from %s", len(e.words), e.dim, path)
log.Printf("[static_embedder] loaded %d words, dim=%d from %s (topN=%d)", len(e.words), e.dim, path, topN)
return nil
}

42
internal/memory/synth_model_test.go
View File

@ -86,3 +86,45 @@ func newSynthEmbedder(t testing.TB, dim int) *StaticEmbedder {
}
return e
}
// Phase 5: #topN 规格裁剪加载——只加载前 N 个词向量，控制常驻内存
func TestStaticEmbedderTopNSpec(t *testing.T) {
path := writeSynthModel(t, 300)
// 解析规格
cleanPath, topN := parseModelSpec(path + "#top5")
if cleanPath != path || topN != 5 {
t.Fatalf("parseModelSpec(#top5) = (%q, %d), want (%q, 5)", cleanPath, topN, path)
}
cleanPath2, topN2 := parseModelSpec(path)
if cleanPath2 != path || topN2 != 0 {
t.Fatalf("parseModelSpec(plain) = (%q, %d), want (%q, 0)", cleanPath2, topN2, path)
}
cleanPath3, topN3 := parseModelSpec(path + "#abc")
if cleanPath3 != path || topN3 != 0 {
t.Fatalf("parseModelSpec(#abc) = (%q, %d), want (%q, 0)", cleanPath3, topN3, path)
}
// 裁剪加载
e := NewStaticEmbedder(path + "#top5")
if !e.Loaded() {
t.Fatal("topN embedder should be loaded")
}
if len(e.words) != 5 {
t.Errorf("expected 5 words loaded with #top5, got %d", len(e.words))
}
}
// Phase 5: 裁剪后向量化仍可用（未命中词走 unkVec 兜底）
func TestStaticEmbedderTopNVectorize(t *testing.T) {
path := writeSynthModel(t, 300)
e := NewStaticEmbedder(path + "#top1")
if !e.Loaded() {
t.Fatal("embedder should be loaded")
}
v := e.Vectorize("天气怎么样")
// 未命中词不应产生空向量（unkVec 兜底）
if len(v) == 0 {
t.Error("vectorize with topN=1 should still produce a vector (unkVec fallback)")
}
}

6
internal/network/monitor.go
View File

@ -57,7 +57,11 @@ func (m *Monitor) Start(ctx context.Context, endpoints []string) {
}
m.mu.Unlock()
ticker := time.NewTicker(m.interval)
interval := m.interval
if interval <= 0 {
interval = 30 * time.Second
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
m.checkAll(ctx)

364
internal/plugins/agentcli/plugin.go
View File

@ -46,8 +46,17 @@ func terminalRunning(t *TerminalSession) bool {
return t.cmd != nil && (t.cmd.ProcessState == nil || !t.cmd.ProcessState.Exited())
}
// terminalWatch 终端提醒规则（由 terminal_watch 工具设置）。
type terminalWatch struct {
interval time.Duration // 固定时间反馈间隔，0 禁用
onExit bool // 命令执行结束提醒（默认 true）
bufferBytes int // 该终端专用缓冲阈值（字节），0 使用全局 notify_bytes
quiet bool // 静默模式：不随输出流通知，仅定时反馈/结束提醒/空闲汇总
}
type TerminalSession struct {
id string
command string
cmd *exec.Cmd
session ptyTerm
mu sync.Mutex
@ -59,8 +68,15 @@ type TerminalSession struct {
done chan struct{}
// 通知节流字段
unreadBytes int // 最近一次通知后积累的未读字节数
lastNotify time.Time // 最近一次通知时间
unreadBytes int // 最近一次通知后积累的未读字节数
lastNotify time.Time // 最近一次通知时间
lastData time.Time // 最近一次读到的数据时间（用于判定输出停止）
lastFeedback time.Time // 最近一次定时反馈时间
backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍
watch terminalWatch // 该终端的提醒规则
// 实时画面推流（terminal_output 事件）
stream bytes.Buffer // 待推送的增量输出，由 readLoop 每 200ms flush 一次
}
func (t *TerminalSession) Write(input string) (int, error) {
@ -85,12 +101,13 @@ func (t *TerminalSession) Close() {
t.mu.Unlock()
close(t.stopCh)
// 先终止进程（各平台实现：Linux 信号 / Windows TerminateProcess，幂等），再释放资源。
// 不能依赖 cmd.Process.Kill()：Windows 后端 cmd.Process 为占位（仅 Pid）。
if t.session != nil {
_ = t.session.Kill()
}
t.session.Close()
<-t.done
if t.cmd != nil && t.cmd.Process != nil {
t.cmd.Process.Kill()
}
}
func (t *TerminalSession) ReadOutput() string {
@ -119,6 +136,17 @@ func (t *TerminalSession) appendOutput(data []byte) {
}
}
t.buf.Write(data)
// 同步追加到实时画面推流缓冲（最大 64KB，超出丢弃最旧部分）
const maxStream = 64 * 1024
if t.stream.Len()+len(data) > maxStream {
excess := t.stream.Len() + len(data) - maxStream
if t.stream.Len() > excess {
t.stream.Next(excess)
} else {
t.stream.Reset()
}
}
t.stream.Write(data)
}
func (t *TerminalSession) IsExpired() bool {
@ -194,8 +222,10 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
}
s.RegisterTool("terminal_create", sdk.ToolDef{
Name: "terminal_create",
Description: "创建一个新的交互式终端会话。返回终端 ID，后续通过此 ID 进行读写操作。适用于运行交互式程序如 vim、ssh、top、nano 等。终端默认 5 分钟后自动关闭，可通过 timeout 参数调整。",
Name: "terminal_create",
Description: "创建一个新的交互式终端会话。返回终端 ID，后续通过此 ID 进行读写操作。适用于运行交互式程序如 vim、ssh、top、nano 等。" +
"通知模式通过 notify 参数选择（默认 exit）：exit=仅命令执行结束后提醒一次；interval=定时反馈（如 interval=30s 每 30 秒反馈一次状态摘要）；" +
"buffer=未读输出积累到指定字节数后提醒（如 buffer=8192）；多个模式用逗号组合（如 interval=30s,buffer=8192）。终端默认 5 分钟后自动关闭，可通过 timeout 参数调整。",
NoMemory: true,
Parameters: map[string]interface{}{
"type": "object",
@ -204,6 +234,10 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
"type": "string",
"description": "要执行的命令（默认 bash）。如需运行特定程序直接传入即可，例如：vim /tmp/test.txt",
},
"notify": map[string]interface{}{
"type": "string",
"description": "通知模式（可选）：exit（默认，命令结束后提醒）；interval=时长（定时反馈，如 30s/1m）；buffer=字节数（缓冲阈值提醒）；可逗号组合",
},
"timeout": map[string]interface{}{
"type": "string",
"description": "终端自动关闭时间，例如 5m, 10m, 30m, 1h（默认 5m）",
@ -249,8 +283,8 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
})
s.RegisterTool("terminal_read", sdk.ToolDef{
Name: "terminal_read",
Description: "读取指定终端的当前屏幕内容。返回自上次读取以来的新输出。如需持续监控请多次调用。",
Name: "terminal_read",
Description: "读取指定终端的输出。mode=new（默认）返回自上次读取以来的新输出并清空缓冲；mode=now 返回终端当前显示的全部屏幕内容（不清空缓冲）。如需持续监控请多次调用。",
NoMemory: true,
Parameters: map[string]interface{}{
"type": "object",
@ -259,9 +293,13 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
"type": "string",
"description": "终端 ID",
},
"mode": map[string]interface{}{
"type": "string",
"description": "读取模式：new（默认，新输出并清空缓冲）或 now（当前屏幕全部内容，不清理）",
},
"clear": map[string]interface{}{
"type": "boolean",
"description": "读取后是否清除缓冲区（默认 true）",
"description": "读取后是否清除缓冲区（默认与 mode 一致：new 清除，now 不清除）",
},
},
"required": []string{"id"},
@ -326,6 +364,48 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
return p.handleList()
})
s.RegisterTool("terminal_watch", sdk.ToolDef{
Name: "terminal_watch",
Description: "为指定终端设置提醒规则，避免长时间运行任务（编译/下载/构建等）的输出造成通知风暴。" +
"可选规则：interval=固定时间反馈（每隔该时长向 agent 反馈一次终端状态摘要）；" +
"on_exit=命令执行结束提醒；buffer_bytes=未读输出积累到该字节数时提醒一次；" +
"quiet=静默模式（抑制随输出流的通知，仅保留定时反馈与结束提醒，推荐长任务使用）。" +
"未提供的字段保持原值，clear=true 清除全部规则。默认 on_exit=true。",
NoMemory: true,
Parameters: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"id": map[string]interface{}{
"type": "string",
"description": "终端 ID，来自 terminal_create 的返回值",
},
"interval": map[string]interface{}{
"type": "string",
"description": "固定时间反馈间隔，如 30s, 1m, 5m（可选，0 禁用）",
},
"on_exit": map[string]interface{}{
"type": "boolean",
"description": "命令执行结束时是否提醒（默认 true）",
},
"buffer_bytes": map[string]interface{}{
"type": "integer",
"description": "未读输出积累阈值（字节），达到后提醒一次（可选，默认全局 2048）",
},
"quiet": map[string]interface{}{
"type": "boolean",
"description": "静默模式：不随输出流通知，仅保留定时反馈与结束提醒（推荐编译/下载等长任务）",
},
"clear": map[string]interface{}{
"type": "boolean",
"description": "清除该终端全部提醒规则（恢复默认行为）",
},
},
"required": []string{"id"},
},
}, func(args map[string]interface{}) (interface{}, error) {
return p.handleWatch(args)
})
p.wg.Add(1)
go p.cleanupLoop(s)
@ -378,18 +458,28 @@ func (p *Plugin) handleCreate(s *sdk.PluginSDK, args map[string]interface{}) (in
cols = uint16(c)
}
// 通知模式：默认 exit（命令执行结束后提醒一次）。
// 支持 interval=30s / buffer=8192 / quiet，可逗号组合。
watch := terminalWatch{onExit: true, quiet: true}
if notifyStr, ok := args["notify"].(string); ok && notifyStr != "" {
watch = parseNotifyMode(notifyStr, watch)
}
term, cmd, err := newCommandPty(command, rows, cols)
if err != nil {
return map[string]interface{}{"error": fmt.Sprintf("创建终端失败: %v", err)}, nil
}
session := &TerminalSession{
id: "",
command: command,
cmd: cmd,
session: term,
createdAt: time.Now(),
timeout: timeout,
stopCh: make(chan struct{}),
done: make(chan struct{}),
watch: watch,
}
p.mu.Lock()
@ -404,15 +494,34 @@ func (p *Plugin) handleCreate(s *sdk.PluginSDK, args map[string]interface{}) (in
log.Printf("[agentcli] created terminal %s: command=%q timeout=%v rows=%d cols=%d", id, command, timeout, rows, cols)
return map[string]interface{}{
"id": id,
"status": "created",
"command": command,
"timeout": timeout.String(),
"rows": rows,
"cols": cols,
"id": id,
"status": "created",
"command": command,
"timeout": timeout.String(),
"rows": rows,
"cols": cols,
"notify_mode": notifyModeString(watch),
}, nil
}
// notifyModeString 输出可读的通知模式描述。
func notifyModeString(w terminalWatch) string {
var parts []string
if w.onExit {
parts = append(parts, "exit")
}
if w.interval > 0 {
parts = append(parts, "interval="+w.interval.String())
}
if w.bufferBytes > 0 {
parts = append(parts, fmt.Sprintf("buffer=%d", w.bufferBytes))
}
if len(parts) == 0 {
return "quiet"
}
return strings.Join(parts, ",")
}
func (p *Plugin) handleWrite(s *sdk.PluginSDK, args map[string]interface{}) (interface{}, error) {
id, _ := args["id"].(string)
if id == "" {
@ -455,13 +564,49 @@ func (p *Plugin) handleWrite(s *sdk.PluginSDK, args map[string]interface{}) (int
}, nil
}
// parseNotifyMode 解析 notify 参数并合并进 watch。
// 支持：exit / quiet / interval=时长 / buffer=字节数，逗号分隔组合。
func parseNotifyMode(s string, base terminalWatch) terminalWatch {
w := base
for _, part := range strings.Split(s, ",") {
part = strings.TrimSpace(part)
if part == "" {
continue
}
kv := strings.SplitN(part, "=", 2)
key := strings.TrimSpace(kv[0])
val := ""
if len(kv) == 2 {
val = strings.TrimSpace(kv[1])
}
switch key {
case "exit":
w.onExit = true
w.quiet = false
case "quiet", "silent":
w.quiet = true
case "interval":
if d, err := time.ParseDuration(val); err == nil && d > 0 {
w.interval = d
}
case "buffer":
var n int
if _, err := fmt.Sscanf(val, "%d", &n); err == nil && n > 0 {
w.bufferBytes = n
}
}
}
return w
}
func (p *Plugin) handleRead(args map[string]interface{}) (interface{}, error) {
id, _ := args["id"].(string)
if id == "" {
return map[string]interface{}{"error": "id is required"}, nil
}
clear := true
mode, _ := args["mode"].(string)
clear := mode != "now"
if v, ok := args["clear"].(bool); ok {
clear = v
}
@ -474,19 +619,29 @@ func (p *Plugin) handleRead(args map[string]interface{}) (interface{}, error) {
}
var output string
session.mu.Lock()
if clear {
output = session.ReadAndClearOutput()
output = session.buf.String()
session.buf.Reset()
// 实时画面推流缓冲同步清空，避免 terminal_output 事件与读取结果重复
session.stream.Reset()
} else {
output = session.ReadOutput()
output = session.buf.String()
}
session.mu.Unlock()
if output == "" {
output = "[终端无新输出]"
if mode == "now" {
output = "[终端当前无屏幕内容]"
} else {
output = "[终端无新输出]"
}
}
return map[string]interface{}{
"status": "ok",
"terminal": id,
"mode": mode,
"output": output,
"running": terminalRunning(session),
"uptime": time.Since(session.createdAt).String(),
@ -550,6 +705,55 @@ func (p *Plugin) handleClose(args map[string]interface{}) (interface{}, error) {
}, nil
}
func (p *Plugin) handleWatch(args map[string]interface{}) (interface{}, error) {
id, _ := args["id"].(string)
if id == "" {
return map[string]interface{}{"error": "id is required"}, nil
}
p.mu.Lock()
session, ok := p.sessions[id]
p.mu.Unlock()
if !ok {
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
}
session.mu.Lock()
if v, ok := args["clear"].(bool); ok && v {
session.watch = terminalWatch{onExit: true}
} else {
if v, ok := args["interval"].(string); ok && v != "" {
if d, err := time.ParseDuration(v); err == nil && d >= 0 {
session.watch.interval = d
}
}
if v, ok := args["on_exit"].(bool); ok {
session.watch.onExit = v
}
if v, ok := args["buffer_bytes"].(float64); ok && v >= 0 {
session.watch.bufferBytes = int(v)
}
if v, ok := args["quiet"].(bool); ok {
session.watch.quiet = v
}
if session.watch.interval == 0 && session.watch.bufferBytes == 0 && !session.watch.quiet {
session.watch.onExit = true
}
}
w := session.watch
session.mu.Unlock()
log.Printf("[agentcli] watch updated for %s: %+v", id, w)
return map[string]interface{}{
"status": "ok",
"terminal": id,
"interval": w.interval.String(),
"on_exit": w.onExit,
"buffer_bytes": w.bufferBytes,
"quiet": w.quiet,
}, nil
}
func (p *Plugin) handleList() (interface{}, error) {
p.mu.Lock()
defer p.mu.Unlock()
@ -571,6 +775,7 @@ func (p *Plugin) handleList() (interface{}, error) {
}
terms = append(terms, termInfo{
ID: t.id,
Command: t.command,
Uptime: time.Since(t.createdAt).Round(time.Second).String(),
ExpiresIn: remaining.Round(time.Second).String(),
Running: running,
@ -598,9 +803,24 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) {
readCh := make(chan readResult, 4)
go p.reader(t, buf, readCh)
// 实时画面推流 ticker：每 200ms 批量发布一次 terminal_output 事件
flushTicker := time.NewTicker(200 * time.Millisecond)
defer flushTicker.Stop()
// 立即发送首次"终端已启动"通知，让 agent 感知存在
s.InjectText("agentcli", "agentcli", fmt.Sprintf("[终端 %s 已启动]", t.id))
t.lastNotify = time.Now()
now := time.Now()
t.mu.Lock()
t.lastNotify = now
t.lastData = now
t.lastFeedback = now
t.mu.Unlock()
// 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低
hardNotifyBytes := 64 * 1024
hardNotifyInterval := 10 * time.Second
// 输出停止判定：超过该时长无新数据则视为输出停止
quietLatency := 2 * time.Second
for {
if t.IsExpired() {
@ -613,16 +833,52 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) {
}
if !terminalRunning(t) {
s.InjectText("agentcli", "agentcli", fmt.Sprintf("[终端 %s 中的进程已退出]", t.id))
if t.watch.onExit {
s.InjectText("agentcli", "agentcli", fmt.Sprintf("[终端 %s 中的命令已执行结束]", t.id))
} else {
s.InjectText("agentcli", "agentcli", fmt.Sprintf("[终端 %s 中的进程已退出]", t.id))
}
p.mu.Lock()
delete(p.sessions, t.id)
p.mu.Unlock()
return
}
// 固定时间反馈：watch.interval > 0 时每隔该时长主动反馈一次状态摘要
t.mu.Lock()
if t.watch.interval > 0 && time.Since(t.lastFeedback) >= t.watch.interval {
t.lastFeedback = time.Now()
t.lastNotify = t.lastFeedback
unread := t.unreadBytes
t.unreadBytes = 0
preview := previewTail(t.buf.String(), 120)
t.mu.Unlock()
s.InjectText("agentcli", "agentcli",
fmt.Sprintf("[终端 %s 定时反馈: 运行中, 期间新输出约 %d 字节]\n%s", t.id, unread, preview))
continue
}
t.mu.Unlock()
select {
case <-t.stopCh:
return
case <-flushTicker.C:
// 批量推送终端实时画面增量（独立 ticker，避免被高密度数据饿死）
var streamData string
t.mu.Lock()
if t.stream.Len() > 0 {
streamData = t.stream.String()
t.stream.Reset()
}
t.mu.Unlock()
if streamData != "" {
s.Publish(&sdk.Event{
Type: sdk.EventTerminalOutput,
Source: "agentcli",
Payload: map[string]interface{}{"terminal_id": t.id, "output": streamData, "running": terminalRunning(t)},
Timestamp: time.Now().UnixMilli(),
})
}
case r := <-readCh:
if r.err != nil {
// 读取错误/EOF → 立即通知（进程可能已结束）
@ -634,32 +890,66 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) {
copy(data, buf[:r.n])
t.appendOutput(data)
// 语义通知：累积未读字节数
// 缓冲阈值通知（仅当 agent 显式选择 buffer 模式，或未读积累达到硬上限）。
// 默认模式（仅 exit 提醒）下不随输出流通知，杜绝通知风暴。
t.mu.Lock()
t.lastData = time.Now()
t.unreadBytes += r.n
needNotify := t.unreadBytes >= p.notifyBytes ||
time.Since(t.lastNotify) >= p.notifyInterval
t.mu.Unlock()
if needNotify {
t.mu.Lock()
preview := t.buf.String()
if len(preview) > 200 {
preview = preview[len(preview)-200:] // 取最新 200 字符
bufThr := t.watch.bufferBytes
if bufThr <= 0 {
bufThr = p.notifyBytes
}
minInterval := p.notifyInterval
if t.watch.interval > 0 {
minInterval = t.watch.interval
}
// 风暴退避：距上次通知不足 1s 说明输出极速，通知间隔翻倍（上限 30s）
if time.Since(t.lastNotify) < time.Second && t.unreadBytes >= bufThr {
if t.backoff == 0 {
t.backoff = minInterval
} else if t.backoff < 30*time.Second {
t.backoff *= 2
if t.backoff > 30*time.Second {
t.backoff = 30 * time.Second
}
}
preview = sanitizePreview(preview)
t.unreadBytes = 0
}
interval := t.backoff + minInterval
isHard := t.watch.bufferBytes <= 0 && t.unreadBytes >= hardNotifyBytes
if isHard && hardNotifyInterval > interval {
interval = hardNotifyInterval
}
need := t.unreadBytes >= bufThr && time.Since(t.lastNotify) >= interval
if need {
t.lastNotify = time.Now()
t.unreadBytes = 0
preview := previewTail(t.buf.String(), 200)
t.mu.Unlock()
s.InjectText("agentcli", "agentcli",
fmt.Sprintf("[终端 %s 有新输出]\n%s", t.id, preview))
} else {
t.mu.Unlock()
s.InjectText("agentcli", "agentcli", fmt.Sprintf("[终端 %s 有新输出]\n%s", t.id, preview))
}
}
case <-time.After(pollInterval):
// 空闲轮询：输出已停止时复位退避
t.mu.Lock()
if t.backoff > 0 && time.Since(t.lastData) >= quietLatency {
t.backoff = 0
}
t.mu.Unlock()
}
}
}
// previewTail 返回 s 末尾最多 n 字符，并转义控制字符保证可读。
func previewTail(s string, n int) string {
if len(s) > n {
s = s[len(s)-n:]
}
return sanitizePreview(s)
}
type readResult struct {
n int
err error

253
internal/plugins/agentcli/plugin_test.go
View File

@ -4,6 +4,9 @@ package agentcli
import (
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
"time"
@ -389,3 +392,253 @@ func TestToolsRegistered(t *testing.T) {
}
}
}
// ——— Phase 6: 通知节流测试（mock 终端 + 捕获注入） ———
type injectCapture struct {
mu sync.Mutex
texts []string
}
func (c *injectCapture) InjectInterruptText(source, channel, text string) {
c.mu.Lock()
c.texts = append(c.texts, text)
c.mu.Unlock()
}
func (c *injectCapture) InjectText(source, channel, text string) {
c.mu.Lock()
c.texts = append(c.texts, text)
c.mu.Unlock()
}
func (c *injectCapture) InjectTextNoMemory(source, channel, text string) {
c.mu.Lock()
c.texts = append(c.texts, text)
c.mu.Unlock()
}
func (c *injectCapture) InjectInputSync(source, channel, text string) string { return "" }
func (c *injectCapture) snapshot() []string {
c.mu.Lock()
defer c.mu.Unlock()
out := make([]string, len(c.texts))
copy(out, c.texts)
return out
}
// mockTerm 可控输出流的假终端：Read 从 data chan 取数据，可模拟进程退出/读取错误
type mockTerm struct {
mu sync.Mutex
data chan []byte
running bool
err error
}
func newMockTerm() *mockTerm {
return &mockTerm{data: make(chan []byte, 16), running: true}
}
func (m *mockTerm) Read(buf []byte) (int, error) {
for {
m.mu.Lock()
err := m.err
running := m.running
m.mu.Unlock()
if err != nil {
return 0, err
}
if !running {
return 0, fmt.Errorf("process exited")
}
select {
case data, ok := <-m.data:
if !ok {
return 0, fmt.Errorf("closed")
}
n := copy(buf, data)
return n, nil
case <-time.After(20 * time.Millisecond):
}
}
}
func (m *mockTerm) WriteString(s string) (int, error) { return len(s), nil }
func (m *mockTerm) Resize(rows, cols uint16) error { return nil }
func (m *mockTerm) Running() bool {
m.mu.Lock()
defer m.mu.Unlock()
return m.running
}
func (m *mockTerm) Kill() error { return nil }
func (m *mockTerm) Close() error { return nil }
func (m *mockTerm) push(data []byte) {
m.data <- data
}
func (m *mockTerm) setRunning(v bool) {
m.mu.Lock()
m.running = v
m.mu.Unlock()
}
func (m *mockTerm) setErr(err error) {
m.mu.Lock()
m.err = err
m.mu.Unlock()
}
func newTestSession(term ptyTerm) *TerminalSession {
return &TerminalSession{
id: "t1",
session: term,
createdAt: time.Now(),
timeout: 10 * time.Minute,
stopCh: make(chan struct{}),
done: make(chan struct{}),
}
}
func startReadLoop(p *Plugin, s *sdk.PluginSDK, t *TerminalSession) {
p.wg.Add(1)
go p.readLoop(t, s)
}
func waitInjected(c *injectCapture, substr string, timeout time.Duration) bool {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
for _, text := range c.snapshot() {
if strings.Contains(text, substr) {
return true
}
}
time.Sleep(20 * time.Millisecond)
}
return false
}
// Phase 6: 持续吐进度时，通知频率显著低于 500ms/条（节流生效）
func TestReadLoopNotifyThrottle(t *testing.T) {
p := New("agentcli")
p.notifyBytes = 2048
p.notifyInterval = 2 * time.Second
capture := &injectCapture{}
sdkInst := sdk.New("agentcli", sdk.SDKConfig{
RegTool: newToolCapture().RegisterTool,
RegStage: func(sdk.Stage, sdk.StageHandler) {},
RegAPI: func(string) error { return nil },
Settings: sdk.NewSettings("agentcli", nil),
})
sdkInst.SetIOInjector(capture)
term := newMockTerm()
ts := newTestSession(term)
startReadLoop(p, sdkInst, ts)
if !waitInjected(capture, "已启动", 2*time.Second) {
t.Fatal("expected startup notification")
}
// 持续以 100B/50ms(=2KB/s) 吐进度 3 秒
stop := make(chan struct{})
go func() {
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
chunk := make([]byte, 100)
for i := range chunk {
chunk[i] = 'x'
}
for {
select {
case <-stop:
return
case <-ticker.C:
term.push(chunk)
}
}
}()
time.Sleep(3 * time.Second)
close(stop)
notifies := 0
for _, text := range capture.snapshot() {
if strings.Contains(text, "有新输出") {
notifies++
}
}
// 3 秒持续输出，500ms/条 的旧行为应有 6 条；节流后 ≤3 条
if notifies > 3 {
t.Errorf("notify throttle ineffective: %d notifies in 3s (expected <=3)", notifies)
}
if notifies == 0 {
t.Error("expected at least one output notification")
}
close(ts.stopCh)
<-ts.done
}
// Phase 6: 进程退出 → 立即通知（两条路径：PTY Read 返回 EOF 走"读取结束"，
// 或 reader 阻塞时顶部 terminalRunning 检测走"进程已退出"）
func TestReadLoopNotifyOnExit(t *testing.T) {
p := New("agentcli")
p.notifyBytes = 2048
p.notifyInterval = 2 * time.Second
capture := &injectCapture{}
sdkInst := sdk.New("agentcli", sdk.SDKConfig{
RegTool: newToolCapture().RegisterTool,
RegStage: func(sdk.Stage, sdk.StageHandler) {},
RegAPI: func(string) error { return nil },
Settings: sdk.NewSettings("agentcli", nil),
})
sdkInst.SetIOInjector(capture)
term := newMockTerm()
ts := newTestSession(term)
startReadLoop(p, sdkInst, ts)
if !waitInjected(capture, "已启动", 2*time.Second) {
t.Fatal("expected startup notification")
}
term.setRunning(false)
gotExit := waitInjected(capture, "进程已退出", 2*time.Second)
gotReadEnd := waitInjected(capture, "读取结束", time.Second)
if !gotExit && !gotReadEnd {
t.Error("expected immediate notification on process exit (either 进程已退出 or 读取结束)")
}
close(ts.stopCh)
}
// Phase 6: 读取错误/EOF → 立即通知
func TestReadLoopNotifyOnReadError(t *testing.T) {
p := New("agentcli")
p.notifyBytes = 2048
p.notifyInterval = 2 * time.Second
capture := &injectCapture{}
sdkInst := sdk.New("agentcli", sdk.SDKConfig{
RegTool: newToolCapture().RegisterTool,
RegStage: func(sdk.Stage, sdk.StageHandler) {},
RegAPI: func(string) error { return nil },
Settings: sdk.NewSettings("agentcli", nil),
})
sdkInst.SetIOInjector(capture)
term := newMockTerm()
ts := newTestSession(term)
startReadLoop(p, sdkInst, ts)
if !waitInjected(capture, "已启动", 2*time.Second) {
t.Fatal("expected startup notification")
}
term.setErr(fmt.Errorf("read timeout"))
if !waitInjected(capture, "读取结束", 3*time.Second) {
t.Error("expected immediate notification on read error")
}
close(ts.stopCh)
<-ts.done
}

245
internal/plugins/agentcli/pty_windows.go
View File

@ -8,208 +8,31 @@ import (
"os/exec"
"strings"
"sync"
"syscall"
"unsafe"
"gitcode.com/JianFeeeee/HomeAgent/internal/ptywin"
)
var (
kernel32 = syscall.NewLazyDLL("kernel32.dll")
procCreatePseudoConsole = kernel32.NewProc("CreatePseudoConsole")
procResizePseudoConsole = kernel32.NewProc("ResizePseudoConsole")
procClosePseudoConsole = kernel32.NewProc("ClosePseudoConsole")
procInitializeProcThreadAttributeList = kernel32.NewProc("InitializeProcThreadAttributeList")
procUpdateProcThreadAttribute = kernel32.NewProc("UpdateProcThreadAttribute")
procDeleteProcThreadAttributeList = kernel32.NewProc("DeleteProcThreadAttributeList")
procCreateProcessW = kernel32.NewProc("CreateProcessW")
procGetExitCodeProcess = kernel32.NewProc("GetExitCodeProcess")
procTerminateProcess = kernel32.NewProc("TerminateProcess")
procCloseHandle = kernel32.NewProc("CloseHandle")
)
const (
procThreadAttributePseudoConsole = 0x16 // PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE (22)
extendedStartupinfoPresent = 0x00080000
createUnicodeEnvironment = 0x00000400
stillActive = 259 // STILL_ACTIVE
)
type coord struct {
x int16
y int16
}
type processInformation struct {
process syscall.Handle
thread syscall.Handle
pid uint32
tid uint32
}
// startupInfoEx 对应 STARTUPINFOEXW：STARTUPINFOW 之后追加 attribute list 指针。
type startupInfoEx struct {
cb uint32
lpReserved *uint16
lpDesktop *uint16
lpTitle *uint16
dwX uint32
dwY uint32
dwXSize uint32
dwYSize uint32
dwXCountChars uint32
dwYCountChars uint32
dwFillAttribute uint32
dwFlags uint32
wShowWindow uint16
cbReserved2 uint16
lpReserved2 *byte
hStdInput syscall.Handle
hStdOutput syscall.Handle
hStdErr syscall.Handle
lpAttributeList uintptr
}
func defaultShell() string { return "cmd.exe" }
// windowsPty 基于 Windows ConPTY（Pseudo Console）的终端后端。
//
// ConPTY 通过 CreatePseudoConsole 创建伪控制台，子进程以
// PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE 挂到伪控制台。宿主侧使用两根
// 管道与伪控制台通信：我们写 inW（输入）、读 outR（输出）。
// windowsPty 基于 internal/ptywin（ConPTY）的终端后端。
type windowsPty struct {
hpc syscall.Handle // 伪控制台句柄
inW *os.File // 我们向伪控制台写输入
outR *os.File // 我们读伪控制台输出
proc syscall.Handle // 子进程句柄
procID int
c *ptywin.ConPty
cmd *exec.Cmd
attrList []byte
closeOnce sync.Once
}
// newCommandPty 创建 ConPTY 并在其上运行命令（cmd.exe /c <command>）。
func newCommandPty(command string, rows, cols uint16) (ptyTerm, *exec.Cmd, error) {
inR, inW, err := os.Pipe()
if err != nil {
return nil, nil, fmt.Errorf("create input pipe: %w", err)
}
outR, outW, err := os.Pipe()
if err != nil {
inR.Close()
inW.Close()
return nil, nil, fmt.Errorf("create output pipe: %w", err)
}
sz := coord{x: int16(cols), y: int16(rows)}
var hpc syscall.Handle
r, _, e := procCreatePseudoConsole.Call(
uintptr(unsafe.Pointer(&sz)),
inW.Fd(),
outR.Fd(),
0,
uintptr(unsafe.Pointer(&hpc)),
)
if r == 0 {
inR.Close()
inW.Close()
outR.Close()
outW.Close()
return nil, nil, fmt.Errorf("CreatePseudoConsole: %v", e)
}
// 初始化 process thread attribute list 并注入伪控制台句柄
attrList, err := buildAttrList(hpc)
if err != nil {
procClosePseudoConsole.Call(uintptr(hpc))
inR.Close()
inW.Close()
outR.Close()
outW.Close()
return nil, nil, err
}
cmdLine := windowsCommandLine(command)
cli, err := syscall.UTF16PtrFromString(cmdLine)
c, err := ptywin.Start(cmdLine, ptywin.ConPtyDimensions(int(cols), int(rows)))
if err != nil {
return nil, nil, err
return nil, nil, fmt.Errorf("conpty start: %v", err)
}
var si startupInfoEx
si.cb = uint32(unsafe.Sizeof(si))
si.lpAttributeList = uintptr(unsafe.Pointer(&attrList[0]))
var pi processInformation
flags := uint32(extendedStartupinfoPresent | createUnicodeEnvironment)
r, _, e = procCreateProcessW.Call(
0, // 应用名
uintptr(unsafe.Pointer(cli)), // 命令行（CreateProcessW 会就地改写，可写 buffer）
0, 0, // 无安全属性
0, // bInheritHandles FALSE
uintptr(flags), // 创建标志
0, // 环境（继承）
0, // 工作目录
uintptr(unsafe.Pointer(&si)),
uintptr(unsafe.Pointer(&pi)),
)
if r == 0 {
procDeleteProcThreadAttributeList.Call(uintptr(unsafe.Pointer(&attrList[0])))
procClosePseudoConsole.Call(uintptr(hpc))
inR.Close()
inW.Close()
outR.Close()
outW.Close()
return nil, nil, fmt.Errorf("CreateProcessW: %v", e)
}
// 子进程无需 pipe 的父侧副本；我们只保留 inW/outR
inR.Close()
outW.Close()
cmdObj := exec.Command("cmd.exe")
cmdObj.Process = &os.Process{Pid: int(pi.pid)}
cmdObj.Process = &os.Process{Pid: c.Pid()}
pt := &windowsPty{
hpc: hpc,
inW: inW,
outR: outR,
proc: pi.process,
procID: int(pi.pid),
cmd: cmdObj,
attrList: attrList,
}
return pt, cmdObj, nil
}
func buildAttrList(hpc syscall.Handle) ([]byte, error) {
var size uintptr
r, _, e := procInitializeProcThreadAttributeList.Call(0, 1, 0, uintptr(unsafe.Pointer(&size)))
if r == 0 || size == 0 {
return nil, fmt.Errorf("InitializeProcThreadAttributeList(size): %v", e)
}
buf := make([]byte, size)
r, _, e = procInitializeProcThreadAttributeList.Call(
uintptr(unsafe.Pointer(&buf[0])),
1,
0,
uintptr(unsafe.Pointer(&size)),
)
if r == 0 {
return nil, fmt.Errorf("InitializeProcThreadAttributeList: %v", e)
}
r, _, e = procUpdateProcThreadAttribute.Call(
uintptr(unsafe.Pointer(&buf[0])),
0,
procThreadAttributePseudoConsole,
uintptr(hpc),
unsafe.Sizeof(hpc),
0,
0,
)
if r == 0 {
procDeleteProcThreadAttributeList.Call(uintptr(unsafe.Pointer(&buf[0])))
return nil, fmt.Errorf("UpdateProcThreadAttribute: %v", e)
}
return buf, nil
return &windowsPty{c: c, cmd: cmdObj}, cmdObj, nil
}
func windowsCommandLine(command string) string {
@ -217,43 +40,24 @@ func windowsCommandLine(command string) string {
}
func (p *windowsPty) Read(buf []byte) (int, error) {
return p.outR.Read(buf)
return p.c.Read(buf)
}
func (p *windowsPty) WriteString(s string) (int, error) {
return p.inW.WriteString(s)
return p.c.Write([]byte(s))
}
func (p *windowsPty) Resize(rows, cols uint16) error {
if p.hpc == 0 {
return fmt.Errorf("pseudo console closed")
}
sz := coord{x: int16(cols), y: int16(rows)}
r, _, e := procResizePseudoConsole.Call(uintptr(p.hpc), uintptr(unsafe.Pointer(&sz)))
if r == 0 {
return fmt.Errorf("ResizePseudoConsole: %v", e)
}
return nil
return p.c.Resize(int(cols), int(rows))
}
func (p *windowsPty) Running() bool {
if p.proc == 0 {
return false
}
var code uint32
r, _, _ := procGetExitCodeProcess.Call(uintptr(p.proc), uintptr(unsafe.Pointer(&code)))
if r == 0 {
// 句柄失效（进程已退出并释放句柄）视为停止
return false
}
return code == stillActive
return p.c != nil && p.c.Running()
}
func (p *windowsPty) Kill() error {
if p.proc != 0 {
procTerminateProcess.Call(uintptr(p.proc), 1)
procCloseHandle.Call(uintptr(p.proc))
p.proc = 0
if p.c != nil {
return p.c.Kill()
}
return nil
}
@ -261,28 +65,15 @@ func (p *windowsPty) Kill() error {
func (p *windowsPty) Close() error {
var errs []string
p.closeOnce.Do(func() {
if p.inW != nil {
if err := p.inW.Close(); err != nil {
if p.c != nil {
if err := p.c.Close(); err != nil {
errs = append(errs, err.Error())
}
p.c = nil
}
if p.outR != nil {
if err := p.outR.Close(); err != nil {
errs = append(errs, err.Error())
}
}
if p.hpc != 0 {
procClosePseudoConsole.Call(uintptr(p.hpc))
p.hpc = 0
}
if len(p.attrList) > 0 {
procDeleteProcThreadAttributeList.Call(uintptr(unsafe.Pointer(&p.attrList[0])))
p.attrList = nil
}
_ = p.Kill()
})
if len(errs) > 0 {
return fmt.Errorf("close: %s", strings.Join(errs, "; "))
}
return nil
}
}

65
internal/plugins/agentcli/pty_windows_test.go Normal file
View File

@ -0,0 +1,65 @@
//go:build windows
package agentcli
import (
"strings"
"testing"
"time"
)
// TestNewCommandPtyConPTY 验证 Windows ConPTY 后端：一次性命令输出可读，
// 交互式会话可写读往返。
func TestNewCommandPtyConPTY(t *testing.T) {
ta, _, err := newCommandPty("cmd.exe /c echo conpty-ok", 24, 80)
if err != nil {
t.Fatalf("once: %v", err)
}
outA := drainFor(ta, 3*time.Second)
if !strings.Contains(string(outA), "conpty-ok") {
t.Fatalf("once output missing echo: %q", string(outA))
}
ta.Close()
tb, _, err := newCommandPty("cmd.exe", 24, 80)
if err != nil {
t.Fatalf("interactive: %v", err)
}
defer tb.Close()
time.Sleep(300 * time.Millisecond)
if _, err := tb.WriteString("echo hi-123\r\n"); err != nil {
t.Fatalf("write: %v", err)
}
outB := drainFor(tb, 3*time.Second)
if !strings.Contains(string(outB), "hi-123") {
t.Fatalf("interactive output missing echo: %q", string(outB))
}
if !tb.Running() {
t.Fatalf("interactive shell should still be running")
}
}
func drainFor(term ptyTerm, dur time.Duration) []byte {
deadline := time.Now().Add(dur)
buf := make([]byte, 4096)
var out []byte
for time.Now().Before(deadline) {
ch := make(chan struct{ N int; E error }, 1)
go func() {
n, e := term.Read(buf)
ch <- struct{ N int; E error }{n, e}
}()
select {
case r := <-ch:
if r.N > 0 {
out = append(out, buf[:r.N]...)
}
if r.E != nil {
return out
}
case <-time.After(500 * time.Millisecond):
return out
}
}
return out
}

8
internal/plugins/cli/plugin.go
View File

@ -58,6 +58,14 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
payload, _ := args["payload"].(string)
if payload != "" {
fmt.Println(payload)
s.Publish(&sdk.Event{
Type: sdk.EventAgentOutput,
Payload: map[string]interface{}{
"content": payload,
"channel": "cli",
"kind": "channel_output",
},
})
}
return map[string]interface{}{"status": "ok"}, nil
})

19
internal/plugins/cmd/plugin.go
View File

@ -159,16 +159,19 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
// Windows 上预置 chcp 65001 确保控制台输出为 UTF-8，避免 GBK 乱码
execCmd := command
// Windows 上必须经 cmd.exe /c 执行（chcp 65001 预置为 UTF-8 输出），
// 直接 exec 会把整条命令当成一个程序路径导致所有命令失败。
var cmd *exec.Cmd
if isWindows {
execCmd = "chcp 65001>nul & " + command
execCmd := "chcp 65001>nul & " + command
cmd = exec.CommandContext(ctx, "cmd.exe", "/d", "/c", execCmd)
} else {
parts := shellUnquote(command)
if len(parts) == 0 {
return map[string]interface{}{"error": "command is required"}, nil
}
cmd = exec.CommandContext(ctx, parts[0], parts[1:]...)
}
parts := shellUnquote(execCmd)
if len(parts) == 0 {
return map[string]interface{}{"error": "command is required"}, nil
}
cmd := exec.CommandContext(ctx, parts[0], parts[1:]...)
if workdir != "" {
cmd.Dir = workdir
}

34
internal/plugins/files/plugin.go
View File

@ -6,6 +6,7 @@ import (
"log"
"os"
"path/filepath"
"runtime"
"sort"
"strings"
"sync"
@ -30,6 +31,8 @@ type Plugin struct {
baseDir string // L0 写前留档根目录（<data>/file_baseline 的父目录），空则禁用
}
var isWindowsBuild = runtime.GOOS == "windows"
func New(name string) *Plugin {
return &Plugin{name: name}
}
@ -178,12 +181,41 @@ func (p *Plugin) resolvePath(userPath string) (string, error) {
return "", fmt.Errorf("resolve path: %w", err)
}
base := filepath.Clean(p.filesDir)
if base != "/" && !strings.HasPrefix(abs, base+string(filepath.Separator)) && abs != base {
if !pathWithinSandbox(abs, base) {
return "", fmt.Errorf("path outside sandbox: %s", userPath)
}
return abs, nil
}
// pathWithinSandbox 判断 abs 是否位于沙箱 base 之内。
// Windows 文件系统大小写不敏感，且卷根目录（如 C:\）应放行全盘路径。
func pathWithinSandbox(abs, base string) bool {
lower := func(s string) string {
if isWindowsBuild {
return strings.ToLower(s)
}
return s
}
abs = filepath.Clean(abs)
base = filepath.Clean(base)
if equalFoldPath(abs, base) {
return true
}
// 卷根沙箱（C:\、D:\ 等）表示整机可访问
if isWindowsBuild && len(base) == 3 && base[1] == ':' && base[2] == '\\' {
return true
}
prefix := lower(base) + string(filepath.Separator)
return strings.HasPrefix(lower(abs), prefix)
}
func equalFoldPath(a, b string) bool {
if isWindowsBuild {
return strings.EqualFold(a, b)
}
return a == b
}
func (p *Plugin) handleRead(args map[string]interface{}) (interface{}, error) {
path, _ := args["path"].(string)
if path == "" {

2263
internal/plugins/webui/dashboard.html
View File

File diff suppressed because it is too large Load Diff

244
internal/plugins/webui/handler.go
View File

@ -26,7 +26,35 @@ var dashboardFS embed.FS
var dashboardHTML string
const loginHTML = `<!DOCTYPE html><html lang="zh-CN"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1.0"><title>HomeAgent Login</title><style>body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;background:#0f172a;color:#e2e8f0;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0}.card{background:#1e293b;border:1px solid #334155;border-radius:12px;padding:28px;width:360px}h1{margin:0 0 16px;font-size:20px;color:#38bdf8}label{display:block;font-size:12px;color:#94a3b8;margin:10px 0 4px}input{width:100%;padding:10px 12px;border-radius:8px;border:1px solid #334155;background:#0f172a;color:#e2e8f0}button{width:100%;margin-top:16px;padding:10px 12px;border:none;border-radius:8px;background:#2563eb;color:#fff;font-weight:600;cursor:pointer}.err{margin-top:12px;color:#fca5a5;font-size:13px}</style></head><body><div class="card"><h1>HomeAgent</h1><form id="login-form"><label>用户名</label><input id="username" autocomplete="username"><label>密码</label><input id="password" type="password" autocomplete="current-password"><button type="submit">登录</button><div id="err" class="err"></div></form></div><script>document.getElementById('login-form').addEventListener('submit',async(e)=>{e.preventDefault();const username=document.getElementById('username').value;const password=document.getElementById('password').value;const err=document.getElementById('err');err.textContent='';const r=await fetch('/api/v1/login',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({username,password})});if(r.ok){location.href='/';return}let data={};try{data=await r.json()}catch(_){}err.textContent=data.error||'登录失败'})</script></body></html>`
const loginHTML = `<!DOCTYPE html><html lang="zh-CN"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1.0"><title>HomeAgent 登录</title><style>
:root{--sakura-300:#ffb3c8;--sakura-400:#ff7fac;--sakura-500:#f33b7c;--frost-300:#88c0d0;--text-primary:#e8e6ee;--text-secondary:#a0a3b5;--text-muted:#6e7284;--bg-primary:#0d0d16;--bg-card:rgba(24,24,38,0.72);--bg-input:rgba(13,13,22,0.6);--border-color:rgba(255,255,255,0.09);--glass-border:rgba(255,255,255,0.12);--glass-blur:20px;--radius-lg:18px;--radius-md:12px;--radius-pill:999px;--shadow-glow:0 0 18px rgba(243,59,124,0.35);--ease-out:cubic-bezier(.22,.61,.36,1)}
[data-theme="light"]{--text-primary:#23252e;--text-secondary:#5b5f73;--text-muted:#9aa0b5;--bg-primary:#f6f3f8;--bg-card:rgba(255,255,255,0.72);--bg-input:rgba(255,255,255,0.8);--border-color:rgba(35,37,46,0.1);--glass-border:rgba(255,255,255,0.75);--shadow-glow:0 0 18px rgba(243,59,124,0.28)}
*{box-sizing:border-box}
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI','PingFang SC','Microsoft YaHei',sans-serif;background:radial-gradient(1200px 800px at 15% 0%,rgba(243,59,124,.22),transparent 55%),radial-gradient(1000px 700px at 90% 10%,rgba(136,192,208,.18),transparent 55%),radial-gradient(900px 600px at 50% 110%,rgba(163,184,255,.14),transparent 60%),var(--bg-primary);background-attachment:fixed;color:var(--text-primary);display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;padding:20px;transition:background .3s,color .2s;overflow:hidden}
body::before{content:'';position:fixed;inset:0;pointer-events:none;background-image:radial-gradient(rgba(255,255,255,.05) 1px,transparent 1px);background-size:28px 28px}
.login-wrap{width:100%;max-width:400px;position:relative;z-index:1}
.login-card{background:var(--bg-card);backdrop-filter:blur(var(--glass-blur)) saturate(1.4);-webkit-backdrop-filter:blur(var(--glass-blur)) saturate(1.4);border:1px solid var(--glass-border);border-radius:var(--radius-lg);padding:36px 32px 28px;box-shadow:0 20px 60px rgba(0,0,0,.45)}
.logo{width:84px;height:84px;margin:0 auto 16px;border-radius:50%;overflow:hidden;border:2px solid rgba(255,255,255,.25);box-shadow:0 8px 24px rgba(243,59,124,.35);background:#F8FAFC;display:flex;align-items:center;justify-content:center}
.logo img{width:100%;height:100%;object-fit:cover;display:block}
h1{margin:0 0 6px;font-size:22px;font-weight:700;text-align:center;letter-spacing:-.01em;background:linear-gradient(120deg,var(--sakura-400),var(--frost-300));-webkit-background-clip:text;background-clip:text;-webkit-text-fill-color:transparent}
.sub{margin:0 0 24px;font-size:12px;color:var(--text-muted);text-align:center}
label{display:block;font-size:11px;color:var(--text-secondary);margin:14px 0 6px;font-weight:500;letter-spacing:.03em}
input{width:100%;padding:11px 14px;border-radius:var(--radius-md);border:1px solid var(--border-color);background:var(--bg-input);color:var(--text-primary);font-size:14px;outline:none;transition:border .15s,box-shadow .15s}
input:focus{border-color:var(--sakura-400);box-shadow:var(--shadow-glow)}
input::placeholder{color:var(--text-muted)}
button{width:100%;margin-top:22px;padding:12px 14px;border:none;border-radius:var(--radius-md);background:linear-gradient(120deg,var(--sakura-500),var(--sakura-400));color:#fff;font-size:14px;font-weight:600;cursor:pointer;letter-spacing:.08em;transition:transform .15s var(--ease-out),box-shadow .2s,filter .2s}
button:hover{transform:translateY(-1px);filter:brightness(1.08);box-shadow:0 8px 24px rgba(243,59,124,.4)}
button:active{transform:translateY(0) scale(.98)}
button:disabled{opacity:.6;cursor:not-allowed;transform:none}
.err{margin-top:14px;color:#ff6b6b;font-size:13px;text-align:center;min-height:18px;transition:opacity .2s}
.foot{margin-top:20px;font-size:11px;color:var(--text-muted);text-align:center}
.foot svg{width:12px;height:12px;vertical-align:-2px}
@media(max-width:480px){.login-card{padding:28px 22px 22px}}
</style></head><body><div class="login-wrap"><div class="login-card"><div class="logo"><img src="/logo.svg" alt="HomeAgent"></div><h1>HomeAgent</h1><p class="sub">智能家居助手控制台</p><form id="login-form"><label>用户名</label><input id="username" autocomplete="username" placeholder="请输入用户名" required><label>密码</label><input id="password" type="password" autocomplete="current-password" placeholder="请输入密码" required><button type="submit" id="submit-btn">登 录</button><div id="err" class="err"></div></form></div><div class="foot">HomeAgent &middot; NapCat Theme</div></div><script>
(function(){var m=window.matchMedia('(prefers-color-scheme: light)');function apply(){document.documentElement.setAttribute('data-theme',m.matches?'light':'dark')}apply();m.addEventListener('change',apply)})();
document.getElementById('login-form').addEventListener('submit',async(e)=>{e.preventDefault();const username=document.getElementById('username').value.trim();const password=document.getElementById('password').value;const err=document.getElementById('err');const btn=document.getElementById('submit-btn');err.textContent='';if(!username||!password){err.textContent='请输入用户名和密码';return}btn.disabled=true;btn.textContent='登录中...';try{const r=await fetch('/api/v1/login',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({username,password})});if(r.ok){location.href='/';return}let data={};try{data=await r.json()}catch(_){}err.textContent=data.error||'登录失败'}catch(_){err.textContent='网络错误,请重试'}finally{btn.disabled=false;btn.textContent='登 录'}});
document.getElementById('password').addEventListener('keydown',function(e){if(e.key==='Enter')document.getElementById('login-form').dispatchEvent(new Event('submit'))});
</script></body></html>`
func init() {
data, err := dashboardFS.ReadFile("dashboard.html")
@ -56,6 +84,7 @@ type Handler struct {
chatMu sync.Mutex
chatHistory []ChatMsg
pendingIdx int // chatHistory 中正在进行的 assistant 消息索引，-1 表示无
cmdMu sync.Mutex
cmdHistory []CmdExec
termMu sync.Mutex
@ -63,11 +92,21 @@ type Handler struct {
}
type ChatMsg struct {
Role string `json:"role"`
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content,omitempty"`
Source string `json:"source,omitempty"`
Time string `json:"time"`
Role string `json:"role"`
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content,omitempty"`
ToolCalls []ChatToolCall `json:"tool_calls,omitempty"`
Source string `json:"source,omitempty"`
Time string `json:"time"`
}
type ChatToolCall struct {
Tool string `json:"tool"`
Name string `json:"name,omitempty"`
Args interface{} `json:"args,omitempty"`
Result interface{} `json:"result,omitempty"`
Status string `json:"status,omitempty"`
Plugin string `json:"plugin,omitempty"`
}
type CmdExec struct {
@ -132,15 +171,53 @@ func NewHandler(s *sdk.PluginSDK) *Handler {
llm: llm,
sessions: make(map[string]time.Time),
termStates: make(map[string]*termState),
pendingIdx: -1,
}
h.loadChatHistory()
if s != nil {
go h.trackToolEvents()
h.subscribeChatEvents()
h.subscribeTerminalStream()
}
return h
}
// subscribeTerminalStream 常驻订阅终端实时画面推流（terminal_output 事件），
// 维护 termStates 的 Running 状态与全量输出缓冲，供 /api/v1/terminals 与前端轮询使用。
func (h *Handler) subscribeTerminalStream() {
if h.sdk == nil {
return
}
h.sdk.Subscribe(sdk.EventTerminalOutput, func(ev *sdk.Event) {
id, _ := ev.Payload["terminal_id"].(string)
if id == "" {
return
}
output, _ := ev.Payload["output"].(string)
running, _ := ev.Payload["running"].(bool)
h.termMu.Lock()
ts, ok := h.termStates[id]
if !ok {
ts = &termState{ID: id, created: time.Now()}
h.termStates[id] = ts
}
ts.Running = running
if output != "" {
const maxTermOutput = 64 * 1024
if len(ts.Output)+len(output) > maxTermOutput {
excess := len(ts.Output) + len(output) - maxTermOutput
if len(ts.Output) > excess {
ts.Output = ts.Output[excess:]
} else {
ts.Output = ""
}
}
ts.Output += output
}
h.termMu.Unlock()
})
}
func (h *Handler) loadChatHistory() {
if h.settings == nil {
return
@ -183,6 +260,9 @@ func (h *Handler) subscribeChatEvents() {
if content == "" {
return
}
h.chatMu.Lock()
h.pendingIdx = -1
h.chatMu.Unlock()
h.addChatMsg(ChatMsg{
Role: "user",
Content: content,
@ -190,9 +270,91 @@ func (h *Handler) subscribeChatEvents() {
Time: time.Unix(ev.Timestamp, 0).Format(time.RFC3339),
})
})
h.sdk.Subscribe(sdk.EventToolCall, func(ev *sdk.Event) {
tool, _ := ev.Payload["tool"].(string)
if tool == "" {
return
}
channel, _ := ev.Payload["channel"].(string)
if channel == "_consolidation_" {
return
}
plugin, _ := ev.Payload["plugin"].(string)
status, _ := ev.Payload["status"].(string)
if status == "" {
status = "ok"
}
tc := ChatToolCall{
Tool: tool,
Name: tool,
Args: ev.Payload["args"],
Result: ev.Payload["result"],
Status: status,
Plugin: plugin,
}
h.chatMu.Lock()
msg := h.pendingAssistantLocked()
if msg == nil {
h.chatHistory = append(h.chatHistory, ChatMsg{Role: "assistant", Time: time.Now().Format(time.RFC3339)})
h.pendingIdx = len(h.chatHistory) - 1
msg = &h.chatHistory[h.pendingIdx]
}
msg.ToolCalls = append(msg.ToolCalls, tc)
h.persistChatLocked()
h.chatMu.Unlock()
})
h.sdk.Subscribe(sdk.EventReasoning, func(ev *sdk.Event) {
content, _ := ev.Payload["content"].(string)
if content == "" {
return
}
channel, _ := ev.Payload["channel"].(string)
if channel == "_consolidation_" {
return
}
h.chatMu.Lock()
msg := h.pendingAssistantLocked()
if msg == nil {
h.chatHistory = append(h.chatHistory, ChatMsg{Role: "assistant", Time: time.Now().Format(time.RFC3339)})
h.pendingIdx = len(h.chatHistory) - 1
msg = &h.chatHistory[h.pendingIdx]
}
msg.ReasoningContent += content
h.persistChatLocked()
h.chatMu.Unlock()
})
h.sdk.Subscribe(sdk.EventAgentOutput, func(ev *sdk.Event) {
content, _ := ev.Payload["content"].(string)
channel, _ := ev.Payload["channel"].(string)
kind, _ := ev.Payload["kind"].(string)
h.chatMu.Lock()
// 输出通道主动输出(output_send__{通道})作为独立气泡,不并入最终回复
if kind == "channel_output" {
h.pendingIdx = -1
h.chatMu.Unlock()
if content == "" {
return
}
h.addChatMsg(ChatMsg{
Role: "assistant",
Content: content,
Source: channel,
Time: time.Unix(ev.Timestamp, 0).Format(time.RFC3339),
})
return
}
if msg := h.pendingAssistantLocked(); msg != nil && content != "" {
msg.Content = content
if channel != "" {
msg.Source = channel
}
h.pendingIdx = -1
h.persistChatLocked()
h.chatMu.Unlock()
return
}
h.pendingIdx = -1
h.chatMu.Unlock()
if content == "" {
return
}
@ -205,6 +367,27 @@ func (h *Handler) subscribeChatEvents() {
})
}
// pendingAssistantLocked 返回 chatHistory 中当前进行中的 assistant 消息（已持有 chatMu）。
// 仅当最后一条是 assistant 且尚未产出最终内容时视为进行中，避免跨轮次误合并。
func (h *Handler) pendingAssistantLocked() *ChatMsg {
if h.pendingIdx < 0 || h.pendingIdx >= len(h.chatHistory) {
return nil
}
msg := &h.chatHistory[h.pendingIdx]
if msg.Role != "assistant" || msg.Content != "" {
return nil
}
return msg
}
func (h *Handler) persistChatLocked() {
if h.settings == nil {
return
}
b, _ := json.Marshal(h.chatHistory)
_ = h.settings.Set("chathistory", string(b))
}
func (h *Handler) handleToolEvent(ev *sdk.Event) {
payload := ev.Payload
tool, _ := payload["tool"].(string)
@ -228,7 +411,21 @@ func (h *Handler) handleToolEvent(ev *sdk.Event) {
case "terminal_create":
id := getStr(args, "id")
if id == "" {
// agent 调用时不知道生成的 id，从工具结果中回填
if res, ok := payload["result"].(map[string]interface{}); ok {
id = getStr(res, "id")
}
}
if id == "" {
break
}
cmd := getStr(args, "command")
if cmd == "" {
if res, ok := payload["result"].(map[string]interface{}); ok {
cmd = getStr(res, "command")
}
}
now := time.Now()
term := &termState{
ID: id,
@ -238,7 +435,13 @@ func (h *Handler) handleToolEvent(ev *sdk.Event) {
created: now,
}
h.termMu.Lock()
h.termStates[id] = term
if old, ok := h.termStates[id]; ok {
old.Command = cmd
old.Running = true
old.created = now
} else {
h.termStates[id] = term
}
if len(h.termStates) > maxTerminals {
for k := range h.termStates {
delete(h.termStates, k)
@ -608,7 +811,7 @@ func (h *Handler) handleAgentAction(w http.ResponseWriter, r *http.Request, agen
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": fmt.Sprintf("%s_requested", action), "agent": string(agentID)})
writeJSON(w, http.StatusNotImplemented, map[string]string{"error": fmt.Sprintf("agent %s action not implemented by supervisor", action)})
}
func (h *Handler) handleMemory(w http.ResponseWriter, r *http.Request) {
@ -905,18 +1108,16 @@ func (h *Handler) addChatMsg(msg ChatMsg) {
h.chatMu.Lock()
h.chatHistory = append(h.chatHistory, msg)
if len(h.chatHistory) > maxChatHistory {
h.chatHistory = h.chatHistory[len(h.chatHistory)-maxChatHistory:]
}
// 只持 latest 50 条做持久化（写放大防护），全量仍保留在内存
if h.settings != nil {
persistLen := len(h.chatHistory)
if persistLen > 50 {
persistLen = 50
drop := len(h.chatHistory) - maxChatHistory
h.chatHistory = h.chatHistory[drop:]
if h.pendingIdx >= 0 {
h.pendingIdx -= drop
if h.pendingIdx < 0 {
h.pendingIdx = -1
}
}
toPersist := h.chatHistory[len(h.chatHistory)-persistLen:]
b, _ := json.Marshal(toPersist)
_ = h.settings.Set("chathistory", string(b))
}
h.persistChatLocked()
h.chatMu.Unlock()
}
@ -968,7 +1169,6 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
return
}
// 带超时的上下文，防止 InjectTextSync 长时间阻塞 HTTP 请求
ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
defer cancel()
@ -1052,8 +1252,9 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
log.Printf("[SSE] client reported Last-Event-ID: %s", lastEventID)
}
subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain"}
subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output"}
var unsubs []func()
var seq int64
for _, t := range subTypes {
t2 := t
unsub := h.sdk.Subscribe(sdk.EventType(t2), func(evt *sdk.Event) {
@ -1062,8 +1263,9 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
log.Printf("[SSE] received tool_call event: tool=%s", toolName)
}
data, _ := json.Marshal(evt)
seq++
select {
case writeCh <- fmt.Sprintf("event: %s\ndata: %s\n", evt.Type, string(data)):
case writeCh <- fmt.Sprintf("id: %d-%d\nevent: %s\ndata: %s\n", evt.Timestamp, seq, evt.Type, string(data)):
if evt.Type == sdk.EventToolCall {
toolName, _ := evt.Payload["tool"].(string)
log.Printf("[SSE] wrote tool_call to writeCh: tool=%s", toolName)

6
internal/plugins/webui/handler_test.go
View File

@ -632,13 +632,13 @@ func TestSettingsAPIFlow(t *testing.T) {
if !strings.Contains(html, "settings-layout") {
t.Fatal("HTML should contain settings-layout class")
}
if !strings.Contains(html, "settings-sidebar") {
t.Fatal("HTML should contain settings-sidebar class")
if !strings.Contains(html, "settings-tabs") {
t.Fatal("HTML should contain settings-tabs class")
}
if !strings.Contains(html, "saveSetting") {
t.Fatal("HTML should contain saveSetting JS function")
}
if !strings.Contains(html, "api('/settings'") {
if !strings.Contains(html, `api("/settings"`) {
t.Fatal("HTML should call api('/settings')")
}
})

1
internal/plugins/webui/plugin.go
View File

@ -83,6 +83,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
Payload: map[string]interface{}{
"content": payload,
"channel": "webui",
"kind": "channel_output",
},
})
}

389
internal/ptywin/conpty.go Normal file
View File

@ -0,0 +1,389 @@
//go:build windows
// +build windows
package ptywin
import (
"context"
"errors"
"fmt"
"unicode/utf16"
"unsafe"
"golang.org/x/sys/windows"
)
var (
modKernel32 = windows.NewLazySystemDLL("kernel32.dll")
fCreatePseudoConsole = modKernel32.NewProc("CreatePseudoConsole")
fResizePseudoConsole = modKernel32.NewProc("ResizePseudoConsole")
fClosePseudoConsole = modKernel32.NewProc("ClosePseudoConsole")
fInitializeProcThreadAttributeList = modKernel32.NewProc("InitializeProcThreadAttributeList")
fUpdateProcThreadAttribute = modKernel32.NewProc("UpdateProcThreadAttribute")
ErrConPtyUnsupported = errors.New("ConPty is not available on this version of Windows")
)
func IsConPtyAvailable() bool {
return fCreatePseudoConsole.Find() == nil &&
fResizePseudoConsole.Find() == nil &&
fClosePseudoConsole.Find() == nil &&
fInitializeProcThreadAttributeList.Find() == nil &&
fUpdateProcThreadAttribute.Find() == nil
}
const (
_STILL_ACTIVE uint32 = 259
_S_OK uintptr = 0
_PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE uintptr = 0x20016
defaultConsoleWidth = 80 // in characters
defaultConsoleHeight = 40 // in characters
)
type _COORD struct {
X, Y int16
}
func (c *_COORD) Pack() uintptr {
return uintptr((int32(c.Y) << 16) | int32(c.X))
}
type _HPCON windows.Handle
type handleIO struct {
handle windows.Handle
}
func (h *handleIO) Read(p []byte) (int, error) {
var numRead uint32 = 0
err := windows.ReadFile(h.handle, p, &numRead, nil)
return int(numRead), err
}
func (h *handleIO) Write(p []byte) (int, error) {
var numWritten uint32 = 0
err := windows.WriteFile(h.handle, p, &numWritten, nil)
return int(numWritten), err
}
func (h *handleIO) Close() error {
return windows.CloseHandle(h.handle)
}
type ConPty struct {
hpc _HPCON
pi *windows.ProcessInformation
ptyIn, ptyOut, cmdIn, cmdOut *handleIO
}
func win32ClosePseudoConsole(hPc _HPCON) {
if fClosePseudoConsole.Find() != nil {
return
}
// this kills the attached process. there is no return value.
fClosePseudoConsole.Call(uintptr(hPc))
}
func win32ResizePseudoConsole(hPc _HPCON, coord *_COORD) error {
if fResizePseudoConsole.Find() != nil {
return fmt.Errorf("ResizePseudoConsole not found")
}
ret, _, _ := fResizePseudoConsole.Call(uintptr(hPc), coord.Pack())
if ret != _S_OK {
return fmt.Errorf("ResizePseudoConsole failed with status 0x%x", ret)
}
return nil
}
func win32CreatePseudoConsole(c *_COORD, hIn, hOut windows.Handle) (_HPCON, error) {
if fCreatePseudoConsole.Find() != nil {
return 0, fmt.Errorf("CreatePseudoConsole not found")
}
var hPc _HPCON
ret, _, _ := fCreatePseudoConsole.Call(
c.Pack(),
uintptr(hIn),
uintptr(hOut),
0,
uintptr(unsafe.Pointer(&hPc)))
if ret != _S_OK {
return 0, fmt.Errorf("CreatePseudoConsole() failed with status 0x%x", ret)
}
return hPc, nil
}
type _StartupInfoEx struct {
startupInfo windows.StartupInfo
attributeList []byte
}
func getStartupInfoExForPTY(hpc _HPCON) (*_StartupInfoEx, error) {
if fInitializeProcThreadAttributeList.Find() != nil {
return nil, fmt.Errorf("InitializeProcThreadAttributeList not found")
}
if fUpdateProcThreadAttribute.Find() != nil {
return nil, fmt.Errorf("UpdateProcThreadAttribute not found")
}
var siEx _StartupInfoEx
siEx.startupInfo.Cb = uint32(unsafe.Sizeof(windows.StartupInfo{}) + unsafe.Sizeof(&siEx.attributeList[0]))
siEx.startupInfo.Flags |= windows.STARTF_USESTDHANDLES
var size uintptr
// first call is to get required size. this should return false.
ret, _, _ := fInitializeProcThreadAttributeList.Call(0, 1, 0, uintptr(unsafe.Pointer(&size)))
siEx.attributeList = make([]byte, size, size)
ret, _, err := fInitializeProcThreadAttributeList.Call(
uintptr(unsafe.Pointer(&siEx.attributeList[0])),
1,
0,
uintptr(unsafe.Pointer(&size)))
if ret != 1 {
return nil, fmt.Errorf("InitializeProcThreadAttributeList: %v", err)
}
ret, _, err = fUpdateProcThreadAttribute.Call(
uintptr(unsafe.Pointer(&siEx.attributeList[0])),
0,
_PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE,
uintptr(hpc),
unsafe.Sizeof(hpc),
0,
0)
if ret != 1 {
return nil, fmt.Errorf("InitializeProcThreadAttributeList: %v", err)
}
return &siEx, nil
}
func createConsoleProcessAttachedToPTY(hpc _HPCON, commandLine, workDir string, env []string) (*windows.ProcessInformation, error) {
cmdLine, err := windows.UTF16PtrFromString(commandLine)
if err != nil {
return nil, err
}
var currentDirectory *uint16
if workDir != "" {
currentDirectory, err = windows.UTF16PtrFromString(workDir)
if err != nil {
return nil, err
}
}
var envBlock *uint16
flags := uint32(windows.EXTENDED_STARTUPINFO_PRESENT)
if env != nil {
flags |= uint32(windows.CREATE_UNICODE_ENVIRONMENT)
envBlock = createEnvBlock(env)
}
siEx, err := getStartupInfoExForPTY(hpc)
if err != nil {
return nil, err
}
var pi windows.ProcessInformation
err = windows.CreateProcess(
nil, // use this if no args
cmdLine,
nil,
nil,
false, // inheritHandle
flags,
envBlock,
currentDirectory,
&siEx.startupInfo,
&pi)
if err != nil {
return nil, err
}
return &pi, nil
}
// createEnvBlock refers to syscall.createEnvBlock in go/src/syscall/exec_windows.go
// Sourced From: https://github.com/creack/pty/pull/155
func createEnvBlock(envv []string) *uint16 {
if len(envv) == 0 {
return &utf16.Encode([]rune("\x00\x00"))[0]
}
length := 0
for _, s := range envv {
length += len(s) + 1
}
length += 1
b := make([]byte, length)
i := 0
for _, s := range envv {
l := len(s)
copy(b[i:i+l], []byte(s))
copy(b[i+l:i+l+1], []byte{0})
i = i + l + 1
}
copy(b[i:i+1], []byte{0})
return &utf16.Encode([]rune(string(b)))[0]
}
// This will only return the first error.
func closeHandles(handles ...windows.Handle) error {
var err error
for _, h := range handles {
if h != windows.InvalidHandle {
if err == nil {
err = windows.CloseHandle(h)
} else {
windows.CloseHandle(h)
}
}
}
return err
}
// Close all open handles and terminate the process.
func (cpty *ConPty) Close() error {
// there is no return code
win32ClosePseudoConsole(cpty.hpc)
return closeHandles(
cpty.pi.Process,
cpty.pi.Thread,
cpty.ptyIn.handle,
cpty.ptyOut.handle,
cpty.cmdIn.handle,
cpty.cmdOut.handle)
}
// Wait for the process to exit and return the exit code. If context is canceled,
// Wait() will return STILL_ACTIVE and an error indicating the context was canceled.
func (cpty *ConPty) Wait(ctx context.Context) (uint32, error) {
var exitCode uint32 = _STILL_ACTIVE
for {
if err := ctx.Err(); err != nil {
return _STILL_ACTIVE, fmt.Errorf("wait canceled: %v", err)
}
ret, _ := windows.WaitForSingleObject(cpty.pi.Process, 1000)
if ret != uint32(windows.WAIT_TIMEOUT) {
err := windows.GetExitCodeProcess(cpty.pi.Process, &exitCode)
return exitCode, err
}
}
}
func (cpty *ConPty) Resize(width, height int) error {
coords := _COORD{
int16(width),
int16(height),
}
return win32ResizePseudoConsole(cpty.hpc, &coords)
}
func (cpty *ConPty) Read(p []byte) (int, error) {
return cpty.cmdOut.Read(p)
}
func (cpty *ConPty) Write(p []byte) (int, error) {
return cpty.cmdIn.Write(p)
}
func (cpty *ConPty) Pid() int {
return int(cpty.pi.ProcessId)
}
// Running 报告子进程是否仍在运行（GetExitCodeProcess == STILL_ACTIVE）。
func (cpty *ConPty) Running() bool {
if cpty.pi == nil || cpty.pi.Process == 0 {
return false
}
var code uint32
err := windows.GetExitCodeProcess(cpty.pi.Process, &code)
if err != nil {
return false
}
return code == _STILL_ACTIVE
}
// Kill 强制终止子进程并关闭进程/线程句柄。
func (cpty *ConPty) Kill() error {
if cpty.pi == nil || cpty.pi.Process == 0 {
return nil
}
if err := windows.TerminateProcess(cpty.pi.Process, 1); err != nil {
return err
}
windows.CloseHandle(cpty.pi.Process)
windows.CloseHandle(cpty.pi.Thread)
cpty.pi.Process = 0
cpty.pi.Thread = 0
return nil
}
type conPtyArgs struct {
coords _COORD
workDir string
env []string
}
type ConPtyOption func(args *conPtyArgs)
func ConPtyDimensions(width, height int) ConPtyOption {
return func(args *conPtyArgs) {
args.coords.X = int16(width)
args.coords.Y = int16(height)
}
}
func ConPtyWorkDir(workDir string) ConPtyOption {
return func(args *conPtyArgs) {
args.workDir = workDir
}
}
func ConPtyEnv(env []string) ConPtyOption {
return func(args *conPtyArgs) {
args.env = env
}
}
// Start a new process specified in `commandLine` and attach a pseudo console using the Windows
// ConPty API. If ConPty is not available, ErrConPtyUnsupported will be returned.
//
// On successful return, an instance of ConPty is returned. You must call Close() on this to release
// any resources associated with the process. To get the exit code of the process, you can call Wait().
func Start(commandLine string, options ...ConPtyOption) (*ConPty, error) {
if !IsConPtyAvailable() {
return nil, ErrConPtyUnsupported
}
args := &conPtyArgs{
coords: _COORD{defaultConsoleWidth, defaultConsoleHeight},
}
for _, opt := range options {
opt(args)
}
var cmdIn, cmdOut, ptyIn, ptyOut windows.Handle
if err := windows.CreatePipe(&ptyIn, &cmdIn, nil, 0); err != nil {
return nil, fmt.Errorf("CreatePipe: %v", err)
}
if err := windows.CreatePipe(&cmdOut, &ptyOut, nil, 0); err != nil {
closeHandles(ptyIn, cmdIn)
return nil, fmt.Errorf("CreatePipe: %v", err)
}
hPc, err := win32CreatePseudoConsole(&args.coords, ptyIn, ptyOut)
if err != nil {
closeHandles(ptyIn, ptyOut, cmdIn, cmdOut)
return nil, err
}
pi, err := createConsoleProcessAttachedToPTY(hPc, commandLine, args.workDir, args.env)
if err != nil {
closeHandles(ptyIn, ptyOut, cmdIn, cmdOut)
win32ClosePseudoConsole(hPc)
return nil, fmt.Errorf("Failed to create console process: %v", err)
}
cpty := &ConPty{
hpc: hPc,
pi: pi,
ptyIn: &handleIO{ptyIn},
ptyOut: &handleIO{ptyOut},
cmdIn: &handleIO{cmdIn},
cmdOut: &handleIO{cmdOut},
}
return cpty, nil
}

1
internal/sdk/events.go
View File

@ -16,5 +16,6 @@ const (
EventReasoning = events.EventReasoning
EventStage = events.EventStage
EventSystem = events.EventSystem
EventTerminalOutput = events.EventTerminalOutput
EventAll = events.EventAll
)

6
internal/supervisor/daemon.go
View File

@ -117,7 +117,11 @@ func (d *Daemon) RegisterAgent(id types.AgentID) {
}
func (d *Daemon) healthLoop() {
ticker := time.NewTicker(d.cfg.Daemon.HeartbeatInterval)
interval := d.cfg.Daemon.HeartbeatInterval
if interval <= 0 {
interval = 30 * time.Second
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {