package adapter

// 本文件是策略四的「审查」半边：把它从准入路径上摘下来的派发器（reviewJob /
// scheduleReviewLocked / runReview）与三重上界（同适配器在飞、每会话窗口、每适配器
// 窗口预算）。解析韧性在 flowgate_parse.go，准入判定仍在 flowgate.go。
//
// 为什么必须是后台：`noise_review` 实测 p50 5,294 ms / p95 6,651 ms / max 9,204 ms
// （`~/cy.log`，9 天 1,900 次），而五路入站里 qq-ws、kook-ws、weixin 长轮询都是
// 「单读 goroutine + 同步派发」—— 同步跑在读循环上等于把消息、心跳、探活回执全排在
// 一次 6 秒的模型调用后面（DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。
//
// # 与修复前的语义差异（哪一条变了、哪一条没变）
//
//   - `rate_limited`（策略三，门口丢）：**一字不改**。超窗消息在准入当次就被拒绝、
//     根本不进待处理集合，也不经过审查。
//   - 上限修剪（策略四的硬部分）：**一字不改**。待处理集合超过 maxMessages 时，最旧的
//     条目仍在**准入当次同步**被挤掉并审计为 "capped"；触发水位仍是上限的 80%。
//   - 审查修剪（策略四的模型部分）：**判据不变，时机变了**。触发条件（≥80% 上限、
//     同一会话每窗口至多一次、返回保留编号、越界/重复编号丢弃）逐条沿用；变的是结论
//     落地时点 —— 它不再决定「触发它的那条消息」是否放行，而是修剪**它当时看到的**
//     那批历史条目（审计为 "review_pruned"）。
//     直接后果：`review_pruned` 不再从门口丢消息（实测原本 41–74 条/日），那些消息
//     现在会进入流水线 —— 完整性↑、噪声↑，是否开口仍由 engage_judge / 冷避 / owner
//     闸判（这三处判据一行未动）。
//
// 为什么这个差异可以接受：design/12 §6 策略四给审查的职责是「保护 loop 不被不重要的
// 消息塞满」的历史积压去噪，不是逐条消息的准入判决；把一次 6 秒的模型调用钉在准入
// 路径上，代价（读循环阻塞 → 消息/心跳/探活全排队）远大于收益（少进 41–74 条噪声/日）。
// 真正决定「进不进 loop」的两道判据 —— 限流与上限 —— 都留在准入路径上，且未改口径。

import (
	"context"
	"time"
)

const (
	// reviewTimeout bounds one background reviewer call. The in-flight slot is held
	// for the whole call, so an unbounded call would also freeze every other
	// conversation's review for that account. 上一版没有上界，最长一次实测 9.2 s；
	// 30 s 留足余量而不至于让一次挂死的调用长期占位。
	reviewTimeout = 30 * time.Second
)

// reviewJob is one scheduled reviewer call: the snapshot of the pending set the
// reviewer is asked about, plus the sequence numbers that identify those entries
// when the answer comes back (the buffer keeps moving while the call is in flight).
type reviewJob struct {
	key  string
	msgs []UnifiedMessage
	seqs []uint64
}

// scheduleReviewLocked reports whether this admission should start a background
// review and, when it should, marks that review as started and returns its job.
//
// 三重上界都在这里（调用方须持有 g.mu）：
//  1. 同一适配器至多一次在飞（reviewInFlight）—— 后台审查有界，不会因为多个群
//     同时越阈而并发起 N 个模型调用；
//  2. 同一会话每 window 至多一次（lastReview）—— 修复前它只在成功分支置位，
//     于是 68% 解析失败时「每窗口一次」退化成「每条消息一次」；
//  3. 整个适配器每 window 至多 reviewBudget 次（reviewHits）—— 与模型耗时无关的
//     硬上界，会话再多也乘不上去。
//
// lastReview 在这里（**派发时**）写入，而不是等调用返回：成功与失败两条路径因此
// 同样被节流，这是「每条入站帧都付一次 6 秒」的正面修复。
func (g *FlowGate) scheduleReviewLocked(st *flowSession, key string, now time.Time) *reviewJob {
	if g.review == nil || g.maxMessages <= 0 || len(st.buffer) < g.reviewAt {
		return nil
	}
	if g.reviewInFlight || g.reviewRecently(st, now) || !g.reviewBudgetLocked(now) {
		return nil
	}
	st.lastReview = now
	g.reviewInFlight = true
	g.reviewHits = append(g.reviewHits, now)
	return &reviewJob{key: key, msgs: queueMessages(st.buffer), seqs: bufferSeqs(st.buffer)}
}

// runReview executes one scheduled review off the admission path and folds its
// verdict back into the conversation's pending set. It is the only place the
// Reviewer is called.
//
// 失败方向（安全侧，与修复前 `err != nil` 分支同口径）：审查失败**既不额外丢弃**
// （失败绝不被解释成「低价值 ⇒ 丢掉」），**也不给任何条目保留特权**（失败绝不被
// 解释成「高价值 ⇒ 提权」）；待处理集合原样保留，只由既有的 FIFO 上限兜底。
// 唯一的代价是这一批噪声没被修剪，而窗口节流保证下一次尝试要等到下个窗口。
func (g *FlowGate) runReview(ctx context.Context, job *reviewJob) {
	rctx, cancel := reviewContext(ctx)
	defer cancel()
	keep, err := g.review(rctx, job.msgs)

	g.mu.Lock()
	g.reviewInFlight = false
	var dropped []queuedMsg
	if err == nil {
		// 会话可能在审查期间被 GC 掉（空闲清理），那时结论无处可落，直接丢弃。
		if st := g.sessions[job.key]; st != nil {
			st.buffer, dropped = applyReviewSeqs(st.buffer, job.seqs, keep)
		}
	}
	g.mu.Unlock()

	// 审计在锁外：钩子由 channelbridge 接（落库/面板），不该占着闸门的锁跑 I/O。
	g.auditDropped(dropped, "review_pruned")
}

// reviewContext detaches the reviewer call from the admission caller's cancellation
// — the inbound handler returns the moment the message is admitted, and on several
// adapters its ctx dies with the frame — while still bounding the call so a hung
// model cannot pin the single in-flight slot forever.
func reviewContext(ctx context.Context) (context.Context, context.CancelFunc) {
	if ctx == nil {
		ctx = context.Background()
	}
	return context.WithTimeout(context.WithoutCancel(ctx), reviewTimeout)
}

// reviewRecently throttles review so it does not run on every message: it fires at
// most once per window per conversation, which is the cost guard for the review model.
func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
	if st == nil || st.lastReview.IsZero() {
		return false
	}
	return now.Sub(st.lastReview) < g.window
}

// reviewBudgetLocked enforces the per-adapter ceiling on reviewer calls started in
// the current window. 调用方须持有 g.mu。
func (g *FlowGate) reviewBudgetLocked(now time.Time) bool {
	g.reviewHits = g.trimHits(g.reviewHits, now)
	return len(g.reviewHits) < g.reviewBudget
}

// applyReviewSeqs keeps only the buffered entries the reviewer chose. Entries are
// matched by sequence number, so:
//   - entries evicted by the cap while the call was in flight simply do not appear
//     in the buffer any more (they were audited as "capped" at eviction time);
//   - entries that arrived **after** the snapshot are never judged by this review
//     and always survive (the reviewer never saw them).
func applyReviewSeqs(buffer []queuedMsg, seqs []uint64, keep []int) (kept, dropped []queuedMsg) {
	reviewed := make(map[uint64]bool, len(seqs))
	for _, s := range seqs {
		reviewed[s] = true
	}
	keepSet := make(map[uint64]bool, len(keep))
	for _, i := range keep {
		if i >= 0 && i < len(seqs) {
			keepSet[seqs[i]] = true
		}
	}
	kept = make([]queuedMsg, 0, len(buffer))
	dropped = make([]queuedMsg, 0)
	for _, e := range buffer {
		if !reviewed[e.seq] || keepSet[e.seq] {
			kept = append(kept, e)
			continue
		}
		dropped = append(dropped, e)
	}
	return kept, dropped
}

// bufferSeqs extracts the sequence numbers of buffered entries in order, so a
// review verdict can be applied to exactly the entries it was asked about.
func bufferSeqs(buffer []queuedMsg) []uint64 {
	out := make([]uint64, len(buffer))
	for i, e := range buffer {
		out[i] = e.seq
	}
	return out
}