fix(examples): a2a/acp 回复闭环 + 会话延续 + 同步注入

入站请求从 InjectInterruptText(202 submitted) 改为 InjectInputSync
同步等待回复，直接返回回复文本；支持 params.session_id 延续多轮
上下文；注册为输出通道让回复有落点。详见 TrueAgent 仓库同名 commit。

example/a2a/plg.json

@@ -2,7 +2,7 @@
   "name": "a2a",
   "name_zh": "A2A 代理通信",
   "name_en": "A2A Agent Communication",
-  "version": "1.1.0",
+  "version": "1.2.0",
   "description": "Agent-to-Agent 协议通信插件，支持双向 A2A 通信：可查询其他 Agent 并回复其请求。提供 HTTP 服务端暴露本 Agent 能力。",
   "author": "HomeAgent",
   "entry": "plugin.so",

example/a2a/plugin.go

@@ -21,15 +21,47 @@ type Plugin struct {
 	srvMu      sync.Mutex
 	server     *http.Server
 	serverAddr string
+
+	// 会话表：session_id → 上下文前缀。A2A 无状态协议下由插件侧维护
+	// 多轮上下文：同 session 的后续请求会把之前的对话拼进注入文本。
+	sessMu    sync.Mutex
+	sessions  map[string]*a2aSession
 }
 
+// a2aSession 记录一个会话的轮次历史，用于延续上下文。
+type a2aSession struct {
+	ID       string
+	History  []string // 轮次文本 [user1, agent1, user2, agent2, ...]
+	LastUsed time.Time
+}
+
+// maxSessionTurns 单会话保留的最大轮次对数（防上下文无限膨胀）。
+const maxSessionTurns = 10
+
+// sessionGCPeriod 会话过期清理周期；超过 2 小时未用的会话回收。
+const sessionGCPeriod = 30 * time.Minute
+
 func (p *Plugin) Name() string { return p.name }
 
 func (p *Plugin) Start(s *sdk.PluginSDK) error {
 	s.SetAutoRestart(true)
 	p.sdk = s
+	p.sessions = make(map[string]*a2aSession)
 	tp := p.name + "_"
 
+	// 注册自身为输出通道：agent 回复 emit 到本通道时有落点，
+	// 且 output_list_channels 可见（agent 能主动向 a2a 会话推送消息）。
+	if err := s.RegisterOutputChannel(p.name, 1, "A2A Agent 互联通道（外部 agent 查询的回复由此返回）", sdk.ChannelDef{}, func(args map[string]interface{}) (interface{}, error) {
+		payload, _ := args["payload"].(string)
+		log.Printf("[%s] channel output: %s", p.name, truncateRunes(payload, 120))
+		return map[string]interface{}{"status": "ok"}, nil
+	}); err != nil {
+		log.Printf("[%s] register output channel: %v", p.name, err)
+	}
+
+	// 会话 GC：后台周期回收长期不用的会话
+	go p.sessionGCLoop()
+
 	s.Settings().RegisterDef(sdk.ConfigDef{
 		Key: "listen", Default: "127.0.0.1:12000",
 		Type: "string", DisplayName: "监听地址",
@@ -114,6 +146,29 @@ func (p *Plugin) Stop() error {
 	return nil
 }
 
+// sessionGCLoop 周期清理超时会话。
+func (p *Plugin) sessionGCLoop() {
+	ticker := time.NewTicker(sessionGCPeriod)
+	defer ticker.Stop()
+	for range ticker.C {
+		p.sessMu.Lock()
+		for id, sess := range p.sessions {
+			if time.Since(sess.LastUsed) > 2*time.Hour {
+				delete(p.sessions, id)
+			}
+		}
+		p.sessMu.Unlock()
+	}
+}
+
+func truncateRunes(s string, n int) string {
+	r := []rune(s)
+	if len(r) <= n {
+		return s
+	}
+	return string(r[:n]) + "..."
+}
+
 func (p *Plugin) stopServer() {
 	p.srvMu.Lock()
 	defer p.srvMu.Unlock()
@@ -191,7 +246,8 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) {
 		ID      string `json:"id"`
 		Method  string `json:"method"`
 		Params  struct {
-			Query   string `json:"query,omitempty"`
+			Query     string `json:"query,omitempty"`
+			SessionID string `json:"session_id,omitempty"`
 			Message *struct {
 				Role  string `json:"role"`
 				Parts []struct {
@@ -215,19 +271,60 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) {
 			}
 			queryText = strings.TrimSpace(queryText)
 		}
-
+		if queryText == "" {
-		// Inject into agent pipeline via interrupt (preempt current processing) or direct input
+			http.Error(w, "query/message.text required", http.StatusBadRequest)
-		if queryText != "" {
+			return
-			p.sdk.InjectInterruptText("a2a", "webui", fmt.Sprintf("[来自A2A Agent的查询]\n%s", queryText))
 		}
 
-		// Respond with task accepted
+		// 会话：调用方可指定 session_id 延续多轮上下文；不指定则新建。
+		sessionID := strings.TrimSpace(req.Params.SessionID)
+		injectText := queryText
+		p.sessMu.Lock()
+		if sessionID != "" {
+			sess := p.sessions[sessionID]
+			if sess == nil {
+				sess = &a2aSession{ID: sessionID, LastUsed: time.Now()}
+				p.sessions[sessionID] = sess
+			}
+			sess.LastUsed = time.Now()
+			// 有历史则把上下文拼在前面（截尾防爆量）
+			if len(sess.History) > 0 {
+				ctxText := strings.Join(sess.History, "\n")
+				injectText = "[对话上下文]\n" + ctxText + "\n[本轮输入]\n" + queryText
+			}
+		} else {
+			sessionID = fmt.Sprintf("a2a_%d", time.Now().UnixNano())
+			p.sessions[sessionID] = &a2aSession{ID: sessionID, LastUsed: time.Now()}
+		}
+		p.sessMu.Unlock()
+
+		// 同步注入：阻塞等待 agent 处理完成拿回复（不再抢占打断、
+		// 也不再回 202 让请求方永远等不到结果）。HTTP 超时由调用方控制。
+		reply := p.sdk.InjectInputSync(p.name, p.name,
+			fmt.Sprintf("[来自A2A Agent的查询 session=%s]\n%s\n[注意] 请直接以文本回复本查询，不要调用 output_send__%s——你的最终文本回复会被系统自动返回给请求方。", sessionID, injectText, p.name))
+
+		// 回复写回会话历史（下一轮作为上下文）
+		p.sessMu.Lock()
+		if sess := p.sessions[sessionID]; sess != nil {
+			sess.History = append(sess.History, "用户: "+queryText, "助手: "+reply)
+			if len(sess.History) > maxSessionTurns*2 {
+				sess.History = sess.History[len(sess.History)-maxSessionTurns*2 :]
+			}
+			sess.LastUsed = time.Now()
+		}
+		p.sessMu.Unlock()
+
 		resp := map[string]interface{}{
 			"jsonrpc": "2.0",
 			"id":      req.ID,
 			"result": map[string]interface{}{
 				"id":     fmt.Sprintf("task_%d", time.Now().UnixNano()),
-				"status": "submitted",
+				"status": "completed",
+				"session_id": sessionID,
+				"message": map[string]interface{}{
+					"role": "agent",
+					"parts": []map[string]string{{"type": "text", "text": reply}},
+				},
 			},
 		}
 		w.Header().Set("Content-Type", "application/json")

example/acp/plg.json

@@ -2,11 +2,15 @@
   "name": "acp",
   "name_zh": "ACP 代理通信",
   "name_en": "ACP Agent Client Protocol",
-  "version": "1.0.0",
+  "version": "1.1.0",
   "description": "Agent Client Protocol 通信插件：充当 ACP 服务端接受其他 Agent 的任务请求，同时提供客户端工具向远程 ACP Agent（如 opencode）发起会话并读取回复",
   "author": "HomeAgent",
   "entry": "plugin.so",
-  "tags": ["acp", "agent", "interop"],
+  "tags": [
+    "acp",
+    "agent",
+    "interop"
+  ],
   "targets": "linux/amd64",
   "outdir": "dist",
   "bundle": true,

example/acp/plugin.go

@@ -34,8 +34,13 @@ type Plugin struct {
 type sessionState struct {
 	ID       string
 	Replying []map[string]interface{}
+	History  []string // 轮次历史 [user, agent, user, agent...]，延续上下文用
+	LastUsed time.Time
 }
 
+// maxSessionTurns 单会话保留的最大轮次对数。
+const maxSessionTurns = 10
+
 func (p *Plugin) Name() string { return p.name }
 
 func (p *Plugin) Start(s *sdk.PluginSDK) error {
@@ -44,6 +49,14 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
 	p.sessions = make(map[string]*sessionState)
 	tp := p.name + "_"
 
+	// 注册自身为输出通道：agent 回复 emit 到本通道时有落点。
+	// （回复主要走同步注入返回，此通道用于 agent 主动 output_send__acp）
+	s.RegisterOutputChannel(p.name, 1, "ACP Agent 互联通道（外部 agent 会话的回复由此返回）", sdk.ChannelDef{}, func(args map[string]interface{}) (interface{}, error) {
+		payload, _ := args["payload"].(string)
+		log.Printf("[%s] channel output: %s", p.name, truncateStr(payload, 120))
+		return map[string]interface{}{"status": "ok"}, nil
+	})
+
 	s.Settings().RegisterDef(sdk.ConfigDef{
 		Key: "listen", Default: "127.0.0.1:12001",
 		Type: "string", DisplayName: "监听地址",
@@ -190,21 +203,56 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		sid := fmt.Sprintf("session_%d", time.Now().UnixNano())
+		// 会话：调用方可指定 session_id 延续多轮；不指定则新建。
+		sid := strings.TrimSpace(req.Params.SessionID)
 		p.mu.Lock()
-		p.sessions[sid] = &sessionState{ID: sid}
+		if sid != "" {
+			if _, exists := p.sessions[sid]; !exists {
+				p.sessions[sid] = &sessionState{ID: sid, LastUsed: time.Now()}
+			}
+		} else {
+			sid = fmt.Sprintf("session_%d", time.Now().UnixNano())
+			p.sessions[sid] = &sessionState{ID: sid, LastUsed: time.Now()}
+		}
+		st := p.sessions[sid]
 		p.mu.Unlock()
 
-		if p.sdk != nil {
+		// 延续上下文
-			p.sdk.InjectInterruptText(p.name, "acp",
+		injectText := text
-				fmt.Sprintf("[来自ACP Agent的请求请求 session %s]\n%s", sid, text))
+		p.mu.Lock()
+		if len(st.History) > 0 {
+			ctxText := strings.Join(st.History, "\n")
+			injectText = "[对话上下文]\n" + ctxText + "\n[本轮输入]\n" + text
 		}
+		p.mu.Unlock()
+
+		// 同步注入等待回复：不抢占打断，完整闭环返回文本。
+		reply := ""
+		if p.sdk != nil {
+			reply = p.sdk.InjectInputSync(p.name, p.name,
+				fmt.Sprintf("[来自ACP Agent的请求 session %s]\n%s\n[注意] 请直接以文本回复本请求，不要调用 output_send__%s——你的最终文本回复会被系统自动返回给请求方。", sid, injectText, p.name))
+		}
+
+		// 写回历史 + 填充 Replying 供 SSE 消费
+		p.mu.Lock()
+		st.History = append(st.History, "用户: "+text, "助手: "+reply)
+		if len(st.History) > maxSessionTurns*2 {
+			st.History = st.History[len(st.History)-maxSessionTurns*2:]
+		}
+		st.LastUsed = time.Now()
+		if reply != "" {
+			st.Replying = append(st.Replying, map[string]interface{}{
+				"type": "reply", "text": reply,
+			})
+		}
+		p.mu.Unlock()
 
 		w.Header().Set("Content-Type", "application/json")
 		json.NewEncoder(w).Encode(map[string]interface{}{
 			"jsonrpc": "2.0", "id": req.ID,
 			"result": map[string]interface{}{
 				"session": map[string]interface{}{"id": sid},
+				"reply":   reply,
 			},
 		})
 
@@ -212,12 +260,17 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) {
 		sid := req.Params.SessionID
 		p.mu.Lock()
 		st := p.sessions[sid]
-		if st != nil && req.Params.Final {
-			st.Replying = append(st.Replying, map[string]interface{}{
-				"type": "reply", "text": "done",
-			})
-		}
 		p.mu.Unlock()
+		if st == nil {
+			http.Error(w, "session not found", http.StatusNotFound)
+			return
+		}
+		if req.Params.Final {
+			// 客户端结束会话：标记并保留历史（后续可再 session/new 续）
+			p.mu.Lock()
+			st.LastUsed = time.Now()
+			p.mu.Unlock()
+		}
 
 		w.Header().Set("Content-Type", "application/json")
 		json.NewEncoder(w).Encode(map[string]interface{}{