DEV-MERGE-LEGACY-BACKEND-2026/internal/academic/dualrun/run.go

package dualrun

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"time"
)

// 阶段 1 的冻结点：与 P0 实测一致的整秒 epoch（2026-03-31 10:00:00 +08:00）。
//
// 口径：两侧同源、**Truncate 到整秒**。理由见 P0 §6 组 2b：`SET timestamp` 连微秒都能冻，
// 不取整就会出现"legacy 给 .123456、target 给 .000"的伪差异。
const DefaultFrozenEpoch int64 = 1774922400

// Options 是 stage1 的运行参数（全部来自 CLI）。
type Options struct {
	// CaseSet：real100 / synthetic / all。
	CaseSet string
	// DryRun：只列 case，不执行。
	DryRun bool
	// OutJSON / OutMD 输出路径。
	OutJSON string
	OutMD   string
	// Idempotency：每条 case 连跑两次比对结论。
	Idempotency bool
	// RealCount 真实基线条数（默认 100）。
	RealCount int
	// PerCat 合成每类条数（默认 5）。
	PerCat int
	// FrozenEpoch 冻结点（默认 DefaultFrozenEpoch）。
	FrozenEpoch int64
	// Now 选样用的"当前时间"（教龄换算用；零值取 time.Now）。
	Now time.Time
}

func (o *Options) normalize() {
	if o.CaseSet == "" {
		o.CaseSet = "all"
	}
	if o.RealCount <= 0 {
		o.RealCount = 100
	}
	if o.PerCat <= 0 {
		o.PerCat = 5
	}
	if o.FrozenEpoch == 0 {
		o.FrozenEpoch = DefaultFrozenEpoch
	}
	if o.Now.IsZero() {
		o.Now = time.Now()
	}
	if o.OutJSON == "" {
		o.OutJSON = "dualrun-stage1.json"
	}
	if o.OutMD == "" {
		o.OutMD = "dualrun-stage1.md"
	}
}

// RunStage1 执行阶段 1。
//
// 前置：调用方已 LoadConfigFromEnv 并传入；本函数内部会 Guard()（违反即 panic）。
//
// 出参：Report（同时已写入 OutJSON / OutMD）；error 仅表示"环境层面无法继续"。
func RunStage1(ctx context.Context, cfg Config, opts Options) (*Report, error) {
	opts.normalize()

	// ---- 守卫 1：host / 库名 / 口令（panic 语义见 Config.Guard）----
	cfg.Guard()

	report := &Report{
		Stage:       "AC-15 stage1 (payroll chain)",
		GeneratedAt: time.Now().Format("2006-01-02 15:04:05"),
		CaseSet:     opts.CaseSet,
		DryRun:      opts.DryRun,
		FrozenEpoch: opts.FrozenEpoch,
		FrozenAt:    time.Unix(opts.FrozenEpoch, 0).In(time.Local).Format("2006-01-02 15:04:05"),
		Guard: map[string]string{
			"host":     maskHost(cfg.Host),
			"database": cfg.Name,
			"user":     cfg.User,
			"口令":       "已用 AC15_DB_PASS 传入（不记录）",
		},
		Assertions: map[string]string{},
		CaseCounts: map[string]int{},
		Summary:    map[Attribution]int{},
	}

	db, err := cfg.Open()
	if err != nil {
		return nil, fmt.Errorf("连接失败：%w", err)
	}
	defer db.Close()

	// ---- 守卫 2：连上后核对 SELECT DATABASE() ----
	if err := AssertDatabase(db, cfg.Name); err != nil {
		return nil, fmt.Errorf("数据源守卫失败（SELECT DATABASE()）：%w", err)
	}

	// ---- 启动静态断言：31 个 routine 里 sysdate( 必须为 0 ----
	nSys, err := AssertNoSysdate(db, cfg.Name)
	if err != nil {
		return nil, err
	}
	report.Assertions["sysdate_count"] = fmt.Sprintf("%d（必须为 0）", nSys)
	report.Assertions["frozen_clock"] = fmt.Sprintf("SET SESSION timestamp=%d；time_zone=+08:00；Truncate(time.Second)", opts.FrozenEpoch)
	report.Assertions["连接模型"] = "每个 case 一个新 PHP 进程（= 新连接）；Go 侧另开只读连接选样"

	refs, err := loadRefs(db, cfg.Name)
	if err != nil {
		return nil, err
	}
	report.Assertions["参照表"] = fmt.Sprintf("payments=%d classify=%d courses=%d teachers=%d rules=%d",
		len(refs.pays), len(refs.classifies), len(refs.courseClass), len(refs.teachers), len(refs.detailByID))

	records, err := loadRecords(db, cfg.Name)
	if err != nil {
		return nil, err
	}
	report.Assertions["record_scan"] = fmt.Sprintf("%d 条（teacherid>0 且有 actual_start_time）", len(records))

	// ---- 组装 case ----
	var cases []Case
	if opts.CaseSet == "real100" || opts.CaseSet == "all" {
		cases = append(cases, BuildRealCases(records, refs, opts.RealCount, opts.Now)...)
	}
	if opts.CaseSet == "synthetic" || opts.CaseSet == "all" {
		syn, notes := BuildSyntheticCases(records, refs, opts.PerCat, opts.Now)
		cases = append(cases, syn...)
		report.MutationNotes = append(report.MutationNotes, notes...)
	}
	if len(cases) == 0 {
		return nil, fmt.Errorf("用例集 %q 组装结果为空", opts.CaseSet)
	}
	for _, c := range cases {
		report.CaseCounts[string(c.Kind)]++
	}

	// ---- dry-run：只列清单 ----
	if opts.DryRun {
		report.DryRunCases = cases
		if err := WriteJSON(report, opts.OutJSON); err != nil {
			return nil, err
		}
		if err := WriteMarkdown(report, opts.OutMD); err != nil {
			return nil, err
		}
		return report, nil
	}

	// ---- 实际执行 ----
	oracle, err := NewOracleRunner(cfg)
	if err != nil {
		return nil, err
	}
	defer oracle.Close()

	frozenAt := time.Unix(opts.FrozenEpoch, 0)
	// 事实依据：factsByRecord 用于第 2 步"复刻判定 vs SP 实返"比对
	factsByRecord := map[string]caseFacts{}
	for _, rec := range records {
		factsByRecord[rec.id] = refs.buildFacts(rec, opts.Now)
	}

	for _, c := range cases {
		// 单 case 失败不中断：记录后继续
		res := runOne(ctx, oracle, refs, factsByRecord, c, frozenAt, opts)
		report.Cases = append(report.Cases, res...)
	}

	// ---- 幂等：**所有 case 各跑第二遍**（抽样会漏，差异是逐 case 的）----
	if opts.Idempotency {
		for i, c := range cases {
			_ = i
			first := collectRows(report.Cases, c.ID)
			again := runOne(ctx, oracle, refs, factsByRecord, c, frozenAt, opts)
			second := collectRows(again, c.ID)
			report.Idempotency = append(report.Idempotency, IdempotencyCheck{
				CaseID:    c.ID,
				First:     first.attr,
				Second:    second.attr,
				Same:      first.attr == second.attr,
				RowsEqual: first.json == second.json,
			})
		}
	}

	// ---- 汇总 ----
	for _, cr := range report.Cases {
		report.Summary[cr.Attribution]++
	}

	if err := WriteJSON(report, opts.OutJSON); err != nil {
		return nil, err
	}
	if err := WriteMarkdown(report, opts.OutMD); err != nil {
		return nil, err
	}
	return report, nil
}

type rowSet struct {
	attr Attribution
	json string
}

func collectRows(results []CaseResult, caseID string) rowSet {
	for _, r := range results {
		if r.CaseID != caseID {
			continue
		}
		b, _ := json.Marshal(r.Rows)
		return rowSet{attr: r.Attribution, json: string(b)}
	}
	return rowSet{}
}

// runOne 跑一条 case（BothSupplement 时内部跑两次）。
func runOne(ctx context.Context, oracle *OracleRunner, refs *refs, facts map[string]caseFacts,
	c Case, frozenAt time.Time, opts Options) []CaseResult {

	variants := []bool{c.IsSupplement}
	if c.BothSupplement {
		variants = []bool{false, true}
	}
	var out []CaseResult
	for _, sup := range variants {
		cc := c
		cc.IsSupplement = sup
		res := CaseResult{
			CaseID:       c.ID,
			Category:     c.Category,
			Kind:         c.Kind,
			RecordID:     c.RecordID,
			IsSupplement: sup,
			Note:         c.Note,
		}
		if c.BothSupplement {
			res.CaseID = fmt.Sprintf("%s/sup=%v", c.ID, sup)
		}

		obs, err := oracle.Run(ctx, cc, frozenAt)
		if err != nil && obs == nil {
			res.Attribution = AttrEnvFailure
			res.Note = joinNote(res.Note, "oracle 进程失败："+err.Error())
			out = append(out, res)
			continue
		}
		res.Oracle = obs

		tgt, terr := ComputeTarget(obs)
		res.Target = tgt
		if terr != nil {
			res.Attribution = AttrEnvFailure
			res.Note = joinNote(res.Note, "target 计算出错："+terr.Error())
			out = append(out, res)
			continue
		}

		f := facts[c.RecordID]
		rows, attr, s1 := Attribute(AttrInput{
			Case:              cc,
			Oracle:            obs,
			Target:            tgt,
			PredictedDetailID: f.rule.id,
			PredictedTier:     tierLabel(f.tier),
		})
		// A24：结算必抛异常，异常本身不是失败；把它记进备注便于复核
		if obs.Exception != nil {
			res.Note = joinNote(res.Note, fmt.Sprintf("oracle 抛出 %s（A24 预期，异常前写入值已读回）", obs.Exception.Message))
		}
		res.Rows = rows
		res.Step1 = s1
		res.Attribution = CaseAttribution(rows, attr)
		out = append(out, res)
	}
	return out
}

func joinNote(a, b string) string {
	if a == "" {
		return b
	}
	return a + "；" + b
}

// SortedCaseIDs 便于测试与调试。
func SortedCaseIDs(cases []Case) []string {
	ids := make([]string, 0, len(cases))
	for _, c := range cases {
		ids = append(ids, c.ID)
	}
	sort.Strings(ids)
	return ids
}