memoryweave/go/internal/selfoptimize/pipeline.go

// 织忆 MemoryWeave — 自动化流程编排引擎
package selfoptimize

import (
	"context"
	"log"
	"sync"
	"time"
)

// ─── 自动化流水线 ──────────────────────────────────────

// Pipeline 自动化流程编排器
type Pipeline struct {
	mu         sync.RWMutex
	handlers   map[string]PipelineHandler
	queue      []PipelineTask
	running    bool
	ctx        context.Context
	cancel     context.CancelFunc
}

// PipelineTask 流程任务
type PipelineTask struct {
	ID       string    `json:"id"`
	Type     string    `json:"type"`     // commit / recall / gap / conflict / consolidate
	Payload  interface{} `json:"payload"`
	CreatedAt time.Time `json:"created_at"`
	Status   string    `json:"status"`    // pending / running / done / failed
}

// PipelineHandler 流程处理器
type PipelineHandler func(task *PipelineTask) error

var Flow = NewPipeline()

func NewPipeline() *Pipeline {
	ctx, cancel := context.WithCancel(context.Background())
	return &Pipeline{
		handlers: make(map[string]PipelineHandler),
		ctx:      ctx,
		cancel:   cancel,
	}
}

// Register 注册流程处理器
func (p *Pipeline) Register(taskType string, handler PipelineHandler) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.handlers[taskType] = handler
}

// Enqueue 加入队列
func (p *Pipeline) Enqueue(taskType string, payload interface{}) string {
	p.mu.Lock()
	defer p.mu.Unlock()

	task := PipelineTask{
		ID:        generateTaskID(taskType),
		Type:      taskType,
		Payload:   payload,
		CreatedAt: time.Now(),
		Status:    "pending",
	}
	p.queue = append(p.queue, task)
	return task.ID
}

// Start 启动自动化循环
func (p *Pipeline) Start() {
	p.mu.Lock()
	if p.running {
		p.mu.Unlock()
		return
	}
	p.running = true
	p.mu.Unlock()

	go p.loop()
	log.Println("[pipeline] 自动化流程引擎启动")
}

// Stop 停止
func (p *Pipeline) Stop() {
	p.cancel()
	p.mu.Lock()
	p.running = false
	p.mu.Unlock()
}

func (p *Pipeline) loop() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-p.ctx.Done():
			return
		case <-ticker.C:
			p.processQueue()
		}
	}
}

func (p *Pipeline) processQueue() {
	p.mu.Lock()
	if len(p.queue) == 0 {
		p.mu.Unlock()
		return
	}

	// 取第一个 pending 任务
	var task *PipelineTask
	var idx int
	for i, t := range p.queue {
		if t.Status == "pending" {
			task = &p.queue[i]
			idx = i
			break
		}
	}
	if task == nil {
		p.mu.Unlock()
		return
	}

	task.Status = "running"
	handler, ok := p.handlers[task.Type]
	p.mu.Unlock()

	if !ok {
		p.mu.Lock()
		p.queue[idx].Status = "failed"
		p.mu.Unlock()
		return
	}

	// 执行
	if err := handler(task); err != nil {
		log.Printf("[pipeline] %s failed: %v", task.ID, err)
		p.mu.Lock()
		p.queue[idx].Status = "failed"
		p.mu.Unlock()
	} else {
		p.mu.Lock()
		p.queue[idx].Status = "done"
		p.mu.Unlock()
	}
}

// Stats 流水线统计
func (p *Pipeline) Stats() map[string]interface{} {
	p.mu.RLock()
	defer p.mu.RUnlock()

	pending, running, done, failed := 0, 0, 0, 0
	for _, t := range p.queue {
		switch t.Status {
		case "pending": pending++
		case "running": running++
		case "done": done++
		case "failed": failed++
		}
	}
	return map[string]interface{}{
		"total":   len(p.queue),
		"pending": pending,
		"running": running,
		"done":    done,
		"failed":  failed,
	}
}

func generateTaskID(taskType string) string {
	return taskType + "-" + time.Now().Format("150405") + "-" + randStr(4)
}

func randStr(n int) string {
	const letters = "abcdefghijklmnopqrstuvwxyz0123456789"
	b := make([]byte, n)
	for i := range b {
		b[i] = letters[time.Now().UnixNano()%int64(len(letters))]
	}
	return string(b)
}

// ─── 5 条自动化流程定义 ─────────────────────────────

// Flow 1: commit → 图谱更新 → 冲突检测 → 被动验证
func RegisterCommitFlow(flow *Pipeline) {
	flow.Register("commit", func(task *PipelineTask) error {
		// Enqueue downstream tasks
		// 1. 图谱更新
		flow.Enqueue("graph_update", task.Payload)
		// 2. 冲突检测
		flow.Enqueue("conflict_scan", task.Payload)
		// 3. 被动验证
		flow.Enqueue("passive_validate", task.Payload)
		return nil
	})
}

// Flow 2: recall → 反馈闭环
func RegisterRecallFlow(flow *Pipeline) {
	flow.Register("recall", func(task *PipelineTask) error {
		// recall 后自动记录用途
		flow.Enqueue("recall_feedback_prompt", task.Payload)
		return nil
	})
}

// Flow 3: gap → 关闭 → 填充
func RegisterGapFlow(flow *Pipeline) {
	flow.Register("gap_fill", func(task *PipelineTask) error {
		// 缺口关闭后通知 Agent 学习
		flow.Enqueue("gap_learn", task.Payload)
		return nil
	})
}

// Flow 4: 修正 → 级联审查
func RegisterCorrectFlow(flow *Pipeline) {
	flow.Register("correct", func(task *PipelineTask) error {
		// 修正记忆后级联检查依赖
		flow.Enqueue("cascade_review", task.Payload)
		return nil
	})
}

// Flow 5: 深度整合 → 自优化报告
func RegisterConsolidateFlow(flow *Pipeline) {
	flow.Register("consolidate", func(task *PipelineTask) error {
		// 整合完成后生成报告
		flow.Enqueue("optimize_report", task.Payload)
		return nil
	})
}