v4 architecture: pipeline stages, SDK, event bus, LLM-driven memory consolidation

- SDK PluginAPI (internal/plugin/sdk/): RegisterTool/RegisterStage/Subscribe/Publish
- EventBus (internal/events/): system-level pub/sub with wildcard support
- StageHost (internal/agent/core/stages.go): 7-stage message pipeline
- Agent core: on_input/pre_action/post_action/before_toolcall/after_toolcall/before_output/after_output
- Plugin Registry: SDK plugin registration and tool routing
- GraphDB.MergeEntities: entity consolidation with relation redirection
- memory_merge tool: allows LLM to merge similar entities
- Consolidation task: heartbeat detects conflicts, enqueues via IO for LLM decision
- _consolidation_ internal channel for system-level memory maintenance
- Comprehensive documentation: ARCHITECTURE.md, PLAN.md, DESIGN.md, README.md
- 54 tests across all packages, all passing

internal/memory/graph.go

@@ -520,6 +520,98 @@ func (g *GraphDB) Introspect() (map[string]interface{}, error) {
 	}, nil
 }
 
+// MergeEntities 合并两个实体：将 sourceName 的所有信息合并到 targetName
+// 1. sourceName 的所有关系重新指向 targetName
+// 2. targetName 的 mention_count 增加 sourceName 的计数
+// 3. sourceName 标记为 merged
+// 返回 (关系的重定向数, error)
+func (g *GraphDB) MergeEntities(sourceName, targetName string) (int, error) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+
+	tx, err := g.db.Begin()
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback()
+
+	var sourceID, targetID int64
+	var sourceCount, targetCount int
+
+	err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", sourceName).Scan(&sourceID, &sourceCount)
+	if err != nil {
+		return 0, fmt.Errorf("source entity '%s' not found: %w", sourceName, err)
+	}
+	err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", targetName).Scan(&targetID, &targetCount)
+	if err != nil {
+		return 0, fmt.Errorf("target entity '%s' not found: %w", targetName, err)
+	}
+
+	if sourceID == targetID {
+		return 0, fmt.Errorf("cannot merge entity with itself")
+	}
+
+	// 重定向 source → target 的关系（作为 source）
+	res, err := tx.Exec(
+		`UPDATE relations SET source_id = ?, updated_at = CURRENT_TIMESTAMP
+		 WHERE source_id = ? AND status = 'active'`,
+		targetID, sourceID,
+	)
+	if err != nil {
+		return 0, err
+	}
+	redirectedSource, _ := res.RowsAffected()
+
+	// 重定向 source → target 的关系（作为 target）
+	res, err = tx.Exec(
+		`UPDATE relations SET target_id = ?, updated_at = CURRENT_TIMESTAMP
+		 WHERE target_id = ? AND status = 'active'`,
+		targetID, sourceID,
+	)
+	if err != nil {
+		return 0, err
+	}
+	redirectedTarget, _ := res.RowsAffected()
+
+	// 删除可能产生的自引用关系
+	_, err = tx.Exec(
+		`DELETE FROM relations
+		 WHERE source_id = target_id AND source_id = ?`,
+		targetID,
+	)
+	if err != nil {
+		return 0, err
+	}
+
+	// 更新 target 的 mention_count
+	_, err = tx.Exec(
+		`UPDATE entities SET mention_count = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
+		targetCount+sourceCount, targetID,
+	)
+	if err != nil {
+		return 0, err
+	}
+
+	// 标记 source 为 merged（改名避免 UNIQUE 冲突）
+	_, err = tx.Exec(
+		`UPDATE entities SET name = ? || '@merged_' || ?,
+			mention_count = 0,
+			updated_at = CURRENT_TIMESTAMP
+		 WHERE id = ?`,
+		sourceName, time.Now().Format("20060102150405"), sourceID,
+	)
+	if err != nil {
+		return 0, err
+	}
+
+	if err := tx.Commit(); err != nil {
+		return 0, err
+	}
+
+	total := int(redirectedSource + redirectedTarget)
+	return total, nil
+}
+
 func (g *GraphDB) Archive(days int) (int, error) {
 	g.mu.Lock()
 	defer g.mu.Unlock()

internal/memory/graph_test.go

@@ -214,6 +214,95 @@ func TestIntrospectHotspots(t *testing.T) {
 	}
 }
 
+func TestMergeEntities(t *testing.T) {
+	g := newTestGraph(t)
+	defer os.Remove(g.dbPath)
+	defer g.Close()
+
+	g.Commit([]Triple{
+		{Subject: "张三", Relation: "喜欢", Object: "编程"},
+		{Subject: "张三", Relation: "居住", Object: "北京"},
+	}, "session", 0)
+
+	g.Commit([]Triple{
+		{Subject: "张先生", Relation: "工作", Object: "字节跳动"},
+	}, "session", 0)
+
+	// 合并前：两个实体各有关联
+	stats, _ := g.Introspect()
+	if stats["entity_count"].(int) != 5 {
+		t.Fatalf("expected 5 entities (张三, 编程, 北京, 张先生, 字节跳动), got %d", stats["entity_count"])
+	}
+
+	// 先增加张先生的 mention_count
+	g.Commit([]Triple{
+		{Subject: "张先生", Relation: "喜欢", Object: "Go"},
+	}, "session", 0)
+
+	n, err := g.MergeEntities("张先生", "张三")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n < 2 {
+		t.Errorf("expected at least 2 redirected relations, got %d", n)
+	}
+
+	// source 应被改名
+	result, err := g.Recall(nil, []string{"张先生"}, 1, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(result.Entities) > 0 {
+		t.Error("张先生 should be merged and hidden")
+	}
+
+	// target 的 mention_count 应合并
+	// 验证 target 还存在（seedEntities 精确查找）
+	result2, err := g.Recall([]string{"张三"}, nil, 1, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, e := range result2.Entities {
+		if e.Name == "张三" {
+			found = true
+			if e.MentionCount < 2 {
+				t.Errorf("expected 张三 mention_count >= 2 after merge, got %d", e.MentionCount)
+			}
+			break
+		}
+	}
+	if !found {
+		t.Error("张三 should still exist after merge")
+	}
+}
+
+func TestMergeEntitiesSelf(t *testing.T) {
+	g := newTestGraph(t)
+	defer os.Remove(g.dbPath)
+	defer g.Close()
+
+	g.Commit([]Triple{
+		{Subject: "张三", Relation: "喜欢", Object: "编程"},
+	}, "session", 0)
+
+	_, err := g.MergeEntities("张三", "张三")
+	if err == nil {
+		t.Error("expected error when merging entity with itself")
+	}
+}
+
+func TestMergeEntitiesNonexistent(t *testing.T) {
+	g := newTestGraph(t)
+	defer os.Remove(g.dbPath)
+	defer g.Close()
+
+	_, err := g.MergeEntities("不存在", "张三")
+	if err == nil {
+		t.Error("expected error for nonexistent source")
+	}
+}
+
 func TestPlaceholders(t *testing.T) {
 	if placeholders(0) != "NULL" {
 		t.Errorf("expected NULL for n=0, got %s", placeholders(0))