memoryweave/go/internal/selfoptimize/executor.go

// 织忆 MemoryWeave — 触发器执行器
package selfoptimize

import (
	"log"
	"sync"
	"time"

	"github.com/xiaoxue/memoryweave/internal/governance"
)

// ─── 触发器执行器 ──────────────────────────────────────

// TriggerExecutor 8 类触发器自动执行循环
type TriggerExecutor struct {
	mu            sync.RWMutex
	triggers      []*TriggerDef
	checkInterval time.Duration
	eventBus      TriggerEventBus
}

// TriggerDef 触发器定义
type TriggerDef struct {
	ID        string        `json:"id"`
	Type      TriggerExType `json:"type"`
	Condition interface{}   `json:"condition"`
	Action    string        `json:"action"` // 触发的流水线任务类型
	LastFired time.Time     `json:"last_fired"`
	Cooldown  time.Duration `json:"cooldown"` // 冷却时间
	Enabled   bool          `json:"enabled"`
	FireCount int           `json:"fire_count"`
}

type TriggerExType string

const (
	TCommitCount   TriggerExType = "commit_count"     // 新增 N 条
	TTimeSince     TriggerExType = "time_since"       // 距上次 N 小时
	TRecallMiss    TriggerExType = "recall_miss"      // 连续 N 次 miss
	TQualityDrop   TriggerExType = "quality_drop"     // quality < threshold
	TGapDetected   TriggerExType = "gap_detected"     // 新缺口
	TConflict      TriggerExType = "conflict_pending" // 待解决冲突
	TVValueLow     TriggerExType = "v_value_low"      // V 值过低
	TDecayCritical TriggerExType = "decay_critical"   // 衰减临界
)

type TriggerEventBus interface {
	OnTrigger(trigger *TriggerDef)
}

// 全局触发器状态
var Executor = &TriggerExecutor{
	triggers:      makeTriggers(),
	checkInterval: 5 * time.Minute,
}

func makeTriggers() []*TriggerDef {
	return []*TriggerDef{
		{ID: "t1", Type: TCommitCount, Condition: 50, Action: "consolidate", Cooldown: 24 * time.Hour, Enabled: true},
		{ID: "t2", Type: TTimeSince, Condition: 48.0, Action: "consolidate", Cooldown: 24 * time.Hour, Enabled: true},
		{ID: "t3", Type: TRecallMiss, Condition: 3, Action: "gap_classify", Cooldown: 30 * time.Minute, Enabled: true},
		{ID: "t4", Type: TQualityDrop, Condition: 0.3, Action: "deprecate_review", Cooldown: 1 * time.Hour, Enabled: true},
		{ID: "t5", Type: TGapDetected, Condition: true, Action: "gap_fill", Cooldown: 5 * time.Minute, Enabled: true},
		{ID: "t6", Type: TConflict, Condition: 3, Action: "conflict_muchen", Cooldown: 10 * time.Minute, Enabled: true},
		{ID: "t7", Type: TVValueLow, Condition: 0.2, Action: "mem_review", Cooldown: 12 * time.Hour, Enabled: true},
		// t8 的 Condition 是「当天废弃条数」阈值（计数器语义，不是率）。
		// 2026-10-26：原先这里写 0.1 但判据函数硬编码 >5 —— 配置值从不被读取，
		// 改配置也改不动行为。现改为 Condition: 5（与硬编码等价，行为不变）+ 判据读 t.Condition。
		{ID: "t8", Type: TDecayCritical, Condition: 5.0, Action: "archive_or_boost", Cooldown: 6 * time.Hour, Enabled: true},
	}
}

// Start 启动触发器检查循环
func (te *TriggerExecutor) Start(flow *Pipeline) {
	if te.eventBus == nil {
		te.eventBus = &defaultEventBus{flow: flow}
	}
	go te.loop()
	log.Println("[executor] 触发器执行器启动 (8 triggers)")
}

func (te *TriggerExecutor) loop() {
	ticker := time.NewTicker(te.checkInterval)
	defer ticker.Stop()

	for range ticker.C {
		te.checkAll()
	}
}

func (te *TriggerExecutor) checkAll() {
	te.mu.RLock()
	triggers := make([]*TriggerDef, len(te.triggers))
	copy(triggers, te.triggers)
	te.mu.RUnlock()

	metrics := Dash.Metrics()

	for _, t := range triggers {
		if !t.Enabled {
			continue
		}
		if time.Since(t.LastFired) < t.Cooldown {
			continue
		}
		if te.evaluate(t, metrics) {
			t.LastFired = time.Now()
			t.FireCount++
			te.eventBus.OnTrigger(t)
			log.Printf("[executor] trigger %s fired (count=%d)", t.ID, t.FireCount)
		}
	}
}

func (te *TriggerExecutor) evaluate(t *TriggerDef, metrics map[string]float64) bool {
	switch t.Type {
	case TCommitCount:
		// 需要 commit 计数（从 LanceDB stats 获取）
		return false // 待集成
	case TTimeSince:
		hours := time.Since(t.LastFired).Hours()
		threshold, _ := t.Condition.(float64)
		return hours >= threshold
	case TRecallMiss:
		// 检查最近的 miss 计数。
		// 2026-10-17：recall_hit_rate 已退役（恒=1、无区分度），改用 recall_usefulness_rate
		// （返回且被使用）作为「召回已产生可用结果、应清除 miss」的信号。
		// rate>0 天然排除 N/A 哨兵（-1>0 为 false）⇒ 无数据时不 fire。
		rate, ok := metrics["recall_usefulness_rate"]
		return ok && rate > 0 && t.LastFired.IsZero()
	case TQualityDrop:
		threshold, _ := t.Condition.(float64)
		// N/A 哨兵（-1，样本不足/越界被拦）不得当「低于阈值」触发 —— 先排除。
		v, ok := metrics["recall_usefulness_rate"]
		return ok && !IsNA(v) && v < threshold
	case TGapDetected:
		// ⚠️ 事故面：gap_closure_rate=-1（样本不足/越界）时 `-1 < 0.5` 恒真 ⇒ 每轮 fire。
		// 必须先判 N/A：无数据 ⇒ 不触发缺口学习。
		v, ok := metrics["gap_closure_rate"]
		return ok && !IsNA(v) && v < 0.5
	case TConflict:
		// 原判据 metrics["auto_resolve_rate"] < 0.5 —— 该值恒为 0，故恒 true，t6 空转。
		// 新判据：人侧数据（画像/心迹/牵挂）有冲突时才 fire。
		return hasProfileOrHeartConflict(metrics)
	case TVValueLow:
		threshold, _ := t.Condition.(float64)
		// N/A 哨兵先排除（同 TQualityDrop）。
		v, ok := metrics["recall_usefulness_rate"]
		return ok && !IsNA(v) && v < threshold
	case TDecayCritical:
		// counter 语义：deprecated_per_day 是「当天废弃条数」，阈值同单位（条/天）。
		threshold, _ := t.Condition.(float64)
		v, ok := metrics["deprecated_per_day"]
		return ok && v > threshold
	}
	return false
}

// hasProfileOrHeartConflict 新判据：人侧数据（画像/心迹/牵挂）存在待仲裁冲突时才 fire。
func hasProfileOrHeartConflict(metrics map[string]float64) bool {
	cd := governance.GetGlobalConflictDetector()
	if cd == nil {
		return false
	}
	for _, c := range cd.ListActive() {
		if c.Status == "pending" {
			return true
		}
	}
	return false
}

// Fire 手动触发
func (te *TriggerExecutor) Fire(triggerID string) bool {
	te.mu.RLock()
	defer te.mu.RUnlock()

	for _, t := range te.triggers {
		if t.ID == triggerID {
			t.LastFired = time.Now()
			t.FireCount++
			te.eventBus.OnTrigger(t)
			return true
		}
	}
	return false
}

// List 返回所有触发器状态
func (te *TriggerExecutor) List() []*TriggerDef {
	te.mu.RLock()
	defer te.mu.RUnlock()
	result := make([]*TriggerDef, len(te.triggers))
	copy(result, te.triggers)
	return result
}

// defaultEventBus 默认实现：触发 → 入队到流水线
type defaultEventBus struct {
	flow *Pipeline
}

func (eb *defaultEventBus) OnTrigger(t *TriggerDef) {
	if eb.flow != nil {
		eb.flow.Enqueue(t.Action, map[string]string{
			"trigger_id": t.ID,
			"action":     t.Action,
		})
	}
}