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 }