Cyrene_v2/internal/services/backthink/selfwake.go

package backthink

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
)

// 本文件是**自我唤醒（自排一次思考）的本体**（design/12 §N C，切片 C1-2）。
//
// 形状（C.0）：她自己发起一次唤醒请求 → 先探**自排专用、容量 1** 的待跑槽
// （满 ⇒ 合并，**不过闸、不扣额度**）→ **槽空才**过窗口闸 → 单 worker 跑一轮
// TriggerSelfWake 思考。
//
// 三条不许破坏的约束（各有探针，见 C.8）：
//  1. **顺序**：探槽 → 过闸 → 入槽，三步在**同一把锁**内（C.8 判据 13）。颠倒成
//     「先过闸、再探槽」，被合并的请求也会白扣一次额度，与「合并不消耗预算」
//     自相矛盾（session.RateLimiter.Allow() **成功即消耗**）。
//  2. **闸前 normalize**（C.8 判据 14）：进闸的永远是已回落默认的值，因为
//     RateLimiter 在 limit<=0 时 Allow() 恒 true 且不记录 ⇒ 未回落的值会把
//     「≤0 回落默认」静默变成「无上限」。
//  3. **容量 1 只限自排轮之间**：至多 1 轮自排在跑 + 至多 1 轮自排在排队；
//     心跳轮与事件轮照旧可与自排轮并发（C1 **不**引入全局回合互斥，G2）。
//
// 自排只保证「她能被叫醒思考」；要不要发出去仍由既有的 decide 级联 + PushGate 决定
// （C.4/C.5/C.7），本文件不碰推送策略。

// 自排的哨兵错误。用 errors.Is 判定；它们被直接返回或用 %w 包装。
var (
	// ErrSelfWakeUnavailable 在服务未 Start 或已 Stop 时返回 —— **绝不假装排上**
	// （红线：不伪造成功）。
	ErrSelfWakeUnavailable = errors.New("backthink: self wake unavailable")
	// ErrSelfWakeNotOwner 在非主人环境下请求自排时返回（工具面是第一道闸，
	// 这里是第二道）。design/12 §C.3 / §C.5 的 denied。
	ErrSelfWakeNotOwner = errors.New("backthink: self wake requires owner environment")
)

// SelfWakeReceipt 是一次自排请求的结果。字段与工具结果一一对应（design/12 §C.1/C.3），
// 因此「被上限拒绝」是她**看得见**的合法业务结果，而不是一个 error。
type SelfWakeReceipt struct {
	// Accepted 报告是否真的排出一轮（合并 = true；超限 = false）。
	Accepted bool
	// Coalesced 报告这次请求与在队的那一轮合并了（不额外产生一轮、不花钱）。
	Coalesced bool
	// Reason 是拒绝/说明文案，可直接显示；成功时为空。
	Reason string
	// Used 是**本窗口已用**额度，在消耗之后取值（合并/拒绝路径取当前已用值，
	// 不因这次请求 +1）。
	Used int
	// Limit 与 WindowSec 是本窗口的口径（已 normalize 的值），供面板与工具面显示。
	Limit     int
	WindowSec int
}

// Auditor 是自排审计的窄接口：三字段与网关的 AuditEvent 同形（gateway/audit.go），
// 因此 *gateway.Server 直接满足它。审计出口是**晚绑定**的（见 sinks.go 的
// SetAuditSink）：网关比 backthink 晚组装。
type Auditor interface {
	AuditEvent(who, what, result string)
}

// 自排审计行的 who 与四个动作。what 用点号小写的稳定标签（与 tool.reject.not_allowed
// 同族）；result 是可直接显示给主人的中文。
const (
	selfWakeAuditWho      = "backthink"
	whatSelfWakeQueued    = "backthink.self_wake.queued"
	whatSelfWakeCoalesced = "backthink.self_wake.coalesced"
	whatSelfWakeCapped    = "backthink.self_wake.capped"
	whatSelfWakeDenied    = "backthink.self_wake.denied"
)

// WakeNow 是她自己排起一轮思考的入口（design/12 §N C.1）。它只负责**排**：
// 探槽、过闸、入槽，或如实返回合并/拒绝的收据；真正的思考由单 worker 跑。
//
// 步骤（顺序不许颠倒，C.1 步骤 1~4）：
//  1. 未 Start / 已 Stop ⇒ ErrSelfWakeUnavailable（不假装排上）；
//  2. 非主人环境 ⇒ ErrSelfWakeNotOwner + denied 审计（**不做事**）；
//  3. 先探容量 1 的待跑槽：满 ⇒ 合并（Accepted=true、Coalesced=true），
//     **不过窗口闸、不扣额度**，在队的那一轮仍用先到的理由；
//  4. 槽空 ⇒ 过闸：拒 ⇒ Accepted=false + capped 审计 + 带 source 的 WARN
//     （Allow() 返回 false 时本就未消耗 ⇒ 拒绝不占额度、不产生轮次）；
//     过 ⇒ 入槽（额度此刻已消耗）+ queued 审计。
//
// ctx 只用于身份环境与日志；待跑的那一轮不持有它（不引入延时/欠账语义，
// 也不让一次请求把调用方的 ctx 生命周期绑进 worker）。
func (b *Backthink) WakeNow(ctx context.Context, req SelfWakeRequest) (SelfWakeReceipt, error) {
	gate, running := b.wakeGate()
	if gate == nil || !running {
		return SelfWakeReceipt{}, ErrSelfWakeUnavailable
	}
	if ctx == nil {
		ctx = context.Background()
	}
	// 第二道身份闸（第一道在工具面，C.3）：非主人环境一律拒绝，并留一条 denied
	// 审计 —— 拒绝要看得见，不许静默。loop.SessionEnvFrom 对未挂载 env 的 ctx 返回
	// 主人默认值（loop 的既有向后兼容口径），因此这条判定与工具面读的是同一个环境。
	if env := loop.SessionEnvFrom(ctx); env.ActorEnv != restrict.ActorOwner {
		b.auditEvent(whatSelfWakeDenied, fmt.Sprintf("非主人环境，拒绝自排（actor=%s）", env.ActorEnv))
		return SelfWakeReceipt{}, ErrSelfWakeNotOwner
	}

	receipt, queued := gate.offer(req)
	b.logSelfWake(req, receipt, queued)
	return receipt, nil
}

// logSelfWake 把一次自排请求落到审计与日志上（四条 what 里的三条；denied 在
// WakeNow 的身份闸处记）。审计失败不改变结果：审计是留痕，不是门。
func (b *Backthink) logSelfWake(req SelfWakeRequest, receipt SelfWakeReceipt, queued bool) {
	switch {
	case receipt.Coalesced:
		b.auditEvent(whatSelfWakeCoalesced,
			fmt.Sprintf("已合并到排队中的那一轮（不额外占额度），理由：%s", req.Reason))
		b.log.Info("backthink self wake coalesced",
			"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
	case receipt.Accepted:
		b.auditEvent(whatSelfWakeQueued, fmt.Sprintf("已排入一轮自主思考，理由：%s", req.Reason))
		b.log.Info("backthink self wake queued",
			"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
	case queued:
		// 不可达：queued 为真时 Accepted 必为真。留着是为了让「排上了却没记审计」
		// 在下一次改动里立刻显形，而不是静默少一条。
		b.log.Warn("backthink self wake queued without audit", "reason", req.Reason)
	default:
		b.auditEvent(whatSelfWakeCapped, receipt.Reason)
		b.log.Warn("backthink self wake capped",
			"reason", req.Reason, "used", receipt.Used, "limit", receipt.Limit,
			"window", b.cfg.SelfWakeWindow.String())
	}
}

// auditEvent 把一条审计写进晚绑定的出口（未接线时静默跳过：审计缺失不阻塞业务）。
func (b *Backthink) auditEvent(what, result string) {
	if a := b.currentAuditor(); a != nil {
		a.AuditEvent(selfWakeAuditWho, what, result)
	}
}

// wakeGate 在生命周期锁下取出自排闸并报告服务是否在跑。未 Start / 已 Stop 时
// running 为 false ⇒ 调用方返回 ErrSelfWakeUnavailable（不假装排上）。
func (b *Backthink) wakeGate() (*selfWakeGate, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.selfWake, b.running
}

// selfWakeGate 是自排的**成本闸 + 待跑槽**（design/12 §N C.1/C.5）。它把三件事
// 锁在一起：容量 1 的待跑槽、每窗口上限的滑窗，以及「探槽 → 过闸 → 入槽」的强制顺序。
//
// 锁序：本闸的互斥量只包住自己的状态；RateLimiter 自带内部锁且绝不回调本闸，
// 因此不存在反向加锁路径（C.1 步骤 4 提到的「无锁序问题」）。
type selfWakeGate struct {
	mu sync.Mutex
	// limiter 是每窗口上限的滑窗（复用既有件 ⇒ 零新依赖、零新机制）。
	// 它拿到手的必须是 normalize 之后的值：limit<=0 时它恒放行且不记录。
	limiter *session.RateLimiter
	limit   int
	window  time.Duration
	// granted 是「本窗口已用」的镜像，**只为上报**（收据的 Used 与审计文案）。
	// 判定始终由 limiter 做；两者用同一个时间源与同一个窗口，因此不会各说各话。
	granted []time.Time
	// pending 是容量 1 的待跑槽。非 nil ⇒ 满了，后来的请求合并进它。
	pending *SelfWakeRequest
	// ready 只用来唤醒 worker（容量 1，非阻塞发送）。它不承载状态：
	// 状态在 pending 上，因此不会出现「收到通知但槽是空的」以外的失配。
	ready chan struct{}
}

// newSelfWakeGate 构建闸。limit/window 必须是**已 normalize** 的值
// （≤0 会得到恒放行的滑窗，见 cfg.go 的 normalize 注释）。
func newSelfWakeGate(limit int, window time.Duration) *selfWakeGate {
	return &selfWakeGate{
		limiter: session.NewRateLimiter(limit, window),
		limit:   limit,
		window:  window,
		ready:   make(chan struct{}, 1),
	}
}

// offer 落实 C.1 步骤 2~4：**先探槽**（满 ⇒ 合并，不过闸、不扣额度）→ **槽空才过闸**
// → 入槽；三步在同一把锁内完成，因此两个并发自排不会同时看到「槽空」而各排一轮。
//
// 返回值 queued 报告是否真排出了一轮（合并 = false）。
func (g *selfWakeGate) offer(req SelfWakeRequest) (SelfWakeReceipt, bool) {
	g.mu.Lock()
	defer g.mu.Unlock()

	if g.pending != nil {
		// 合并路径**一次都不调用** Allow() —— 它成功即消耗，碰一下就是白扣额度。
		return g.receiptLocked(true, true, ""), false
	}
	if !g.limiter.Allow() {
		used := g.usedLocked()
		receipt := SelfWakeReceipt{
			Used:      used,
			Limit:     g.limit,
			WindowSec: int(g.window / time.Second),
			Reason: fmt.Sprintf("已达自我唤醒上限（本窗口 %d/%d，窗口 %s），本轮未排队",
				used, g.limit, g.window),
		}
		return receipt, false
	}

	g.granted = append(g.granted, time.Now())
	queued := req
	g.pending = &queued
	select {
	case g.ready <- struct{}{}:
	default: // 已有一条通知在等：worker 取完槽后会再进内层循环，不会漏。
	}
	return g.receiptLocked(true, false, ""), true
}

// take 取走待跑槽（worker 用）。槽空 ⇒ ok=false，调用方回到等待。
func (g *selfWakeGate) take() (SelfWakeRequest, bool) {
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.pending == nil {
		return SelfWakeRequest{}, false
	}
	req := *g.pending
	g.pending = nil
	return req, true
}

// pendingDepth 报告待跑槽是否占用（探针用：判据 13 要确认合并、14 要确认槽已空）。
func (g *selfWakeGate) pendingDepth() int {
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.pending == nil {
		return 0
	}
	return 1
}

// receiptLocked 组装一张收据（调用方必须持有 g.mu）。Used 在**消耗之后**取值。
func (g *selfWakeGate) receiptLocked(accepted, coalesced bool, reason string) SelfWakeReceipt {
	return SelfWakeReceipt{
		Accepted:  accepted,
		Coalesced: coalesced,
		Reason:    reason,
		Used:      g.usedLocked(),
		Limit:     g.limit,
		WindowSec: int(g.window / time.Second),
	}
}

// usedLocked 报告本窗口已用额度并顺手丢弃过期的记录（调用方必须持有 g.mu）。
//
// 它是**只读口径的镜像**：与 limiter 用同一个时间源（time.Now）与同一个窗口，
// 因此报了「已用几档」，而不参与任何判定。
func (g *selfWakeGate) usedLocked() int {
	cutoff := time.Now().Add(-g.window)
	keep := 0
	for _, t := range g.granted {
		if t.After(cutoff) {
			g.granted[keep] = t
			keep++
		}
	}
	g.granted = g.granted[:keep]
	return keep
}

// selfWakeLoop 是自排 worker（design/12 §N C.1）：**单 goroutine** ⇒ 至多一轮自排在跑；
// 待跑槽容量 1 ⇒ 至多一轮自排在排队。两者合起来就是「排一次只跑一轮 + 最多额外欠 1 轮」。
//
// 它不做延时：槽里的请求立刻跑（重启不产生补跑风暴）。ctx 取消（Stop）时退出；
// 槽里若还留着一轮未跑，就地丢弃——没有持久化，也就没有欠账。
func (b *Backthink) selfWakeLoop(ctx context.Context) {
	defer b.wg.Done()
	gate := b.selfWake
	if gate == nil {
		return
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-gate.ready:
		}
		// 内层循环把当时可取的槽位全部取走：合并语义保证它至多取到一轮，
		// 但循环写法让「取到空槽」成为正常出口，而不是靠时序假设。
		for {
			req, ok := gate.take()
			if !ok {
				break
			}
			b.runSelfWake(ctx, req)
		}
	}
}

// runSelfWake 跑一轮自排思考：折算走与事件同一条链（selfWakeTrigger → foldDeclared →
// escalateImport），随后完全复用既有的 think（记录、判定、推送、落库）。
func (b *Backthink) runSelfWake(ctx context.Context, req SelfWakeRequest) {
	b.log.Info("backthink self wake round",
		"reason", req.Reason, "declared", req.Declared.String())
	b.think(ctx, selfWakeTrigger(req))
}