chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 定义运行时规格解析契约与审计模型 (#254)

Browse Source
This commit is contained in:
chengma
2026-08-17 16:53:11 +08:00
parent b11a586527
commit 12414658d0
8 changed files with 736 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin/repository/mysql_db.go
+108 -2
View File
@@ -20,7 +20,7 @@ import (
"cmautobuy/admin/spec"
)
const mysqlSchemaVersion = 25
const mysqlSchemaVersion = 26
// OpenMySQL 打开生产 MySQL 8 数据库。错误信息绝不包含完整 DSN 或密码。
func OpenMySQL(cfg config.DatabaseConfig) (*sql.DB, error) {
@@ -724,10 +724,71 @@ func MigrateMySQL(db *sql.DB) error {
if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 25, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
return fmt.Errorf("记录 MySQL schema v25 失败: %w", err)
}
current = 25
}
if current < 26 {
if err := migrateMySQLV26(db); err != nil {
return fmt.Errorf("执行 MySQL schema v26 失败: %w", err)
}
if err := checkMySQLV26Shape(db); err != nil {
return fmt.Errorf("MySQL schema v26 自检失败，未记录版本: %w", err)
}
if _, err := db.Exec(`INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)`, 26, time.Now().UTC().Format(time.RFC3339Nano)); err != nil {
return fmt.Errorf("记录 MySQL schema v26 失败: %w", err)
}
}
return CheckMySQLSchema(db)
}
// migrateMySQLV26 建立采购运行时规格解析审计。一个记录同时保存候选观察和最终决策，
// 不覆盖 PDD 商品主数据，也不通过外键阻止任务的既有硬删除流程。
func migrateMySQLV26(db *sql.DB) error {
_, err := db.Exec(`CREATE TABLE IF NOT EXISTS purchase_spec_resolutions (
resolution_id VARCHAR(191) COLLATE utf8mb4_bin PRIMARY KEY,
task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
attempt_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
client_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
task_version BIGINT NOT NULL,
pdd_goods_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL,
original_options_json JSON NOT NULL,
selected_color VARCHAR(191) NOT NULL,
target_size VARCHAR(191) NOT NULL,
candidates_json JSON NOT NULL,
candidate_snapshot_hash CHAR(64) COLLATE utf8mb4_bin NOT NULL,
request_hash CHAR(64) COLLATE utf8mb4_bin NOT NULL,
observed_at VARCHAR(35) NOT NULL,
outcome VARCHAR(16) NOT NULL DEFAULT 'pending',
decision_source VARCHAR(16) NULL,
chosen_candidate_id VARCHAR(16) COLLATE utf8mb4_bin NULL,
resolved_options_json JSON NULL,
confidence_bps INT NULL,
reason VARCHAR(500) NULL,
provider_id VARCHAR(191) COLLATE utf8mb4_bin NULL,
source_model VARCHAR(191) NULL,
config_fingerprint CHAR(64) COLLATE utf8mb4_bin NULL,
rules_version VARCHAR(32) NULL,
prompt_version VARCHAR(32) NULL,
created_at VARCHAR(35) NOT NULL,
decided_at VARCHAR(35) NULL,
UNIQUE KEY uq_purchase_spec_resolution_identity (task_id,attempt_id,candidate_snapshot_hash),
KEY idx_purchase_spec_resolution_task (task_id,created_at DESC,resolution_id),
KEY idx_purchase_spec_resolution_outcome (outcome,created_at,resolution_id),
CONSTRAINT chk_purchase_spec_resolution_task_version CHECK (task_version > 0),
CONSTRAINT chk_purchase_spec_resolution_options CHECK (
JSON_TYPE(original_options_json)='OBJECT' AND JSON_LENGTH(original_options_json) BETWEEN 1 AND 16 AND
JSON_TYPE(candidates_json)='ARRAY' AND JSON_LENGTH(candidates_json) BETWEEN 1 AND 100 AND
(resolved_options_json IS NULL OR JSON_TYPE(resolved_options_json)='OBJECT')),
CONSTRAINT chk_purchase_spec_resolution_outcome CHECK (outcome IN ('pending','matched','uncertain','rejected','failed')),
CONSTRAINT chk_purchase_spec_resolution_source CHECK (decision_source IS NULL OR decision_source IN ('rule','ai','reused')),
CONSTRAINT chk_purchase_spec_resolution_confidence CHECK (confidence_bps IS NULL OR confidence_bps BETWEEN 0 AND 10000),
CONSTRAINT chk_purchase_spec_resolution_decision CHECK (
(outcome='pending' AND decided_at IS NULL AND decision_source IS NULL AND chosen_candidate_id IS NULL AND resolved_options_json IS NULL) OR
(outcome='matched' AND decided_at IS NOT NULL AND decision_source IS NOT NULL AND chosen_candidate_id IS NOT NULL AND resolved_options_json IS NOT NULL AND reason IS NOT NULL) OR
(outcome IN ('uncertain','rejected','failed') AND decided_at IS NOT NULL AND chosen_candidate_id IS NULL AND resolved_options_json IS NULL AND reason IS NOT NULL))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`)
return err
}
// migrateMySQLV25 在现有档口入库码业务表上增加后台批次状态，不另建批次表。
func migrateMySQLV25(db *sql.DB) error {
for _, column := range []struct{ name, ddl string }{
@@ -2185,6 +2246,7 @@ func CheckMySQLSchema(db *sql.DB) error {
"ai_provider_configs", "ai_provider_audits",
"ai_spec_match_decisions",
"ai_match_batches", "ai_match_batch_items",
"purchase_spec_resolutions",
}
if err := checkMySQLSchema(db, mysqlRequiredTables); err != nil {
return err
@@ -2255,7 +2317,51 @@ func CheckMySQLSchema(db *sql.DB) error {
if err := checkMySQLV24Shape(db); err != nil {
return err
}
return checkMySQLV25Shape(db)
if err := checkMySQLV25Shape(db); err != nil {
return err
}
return checkMySQLV26Shape(db)
}
func checkMySQLV26Shape(db *sql.DB) error {
if err := checkMySQLSchema(db, []string{"purchase_spec_resolutions"}); err != nil {
return err
}
for _, name := range []string{
"resolution_id", "task_id", "attempt_id", "client_id", "task_version", "pdd_goods_id",
"original_options_json", "selected_color", "target_size", "candidates_json",
"candidate_snapshot_hash", "request_hash", "observed_at", "outcome", "decision_source",
"chosen_candidate_id", "resolved_options_json", "confidence_bps", "reason", "provider_id",
"source_model", "config_fingerprint", "rules_version", "prompt_version", "created_at", "decided_at",
} {
exists, err := mysqlColumnExists(db, "purchase_spec_resolutions", name)
if err != nil || !exists {
return fmt.Errorf("采购运行时规格解析字段 %s 缺失: %v", name, err)
}
}
for _, item := range []struct{ kind, name string }{
{"index", "uq_purchase_spec_resolution_identity"},
{"index", "idx_purchase_spec_resolution_task"},
{"index", "idx_purchase_spec_resolution_outcome"},
{"constraint", "chk_purchase_spec_resolution_task_version"},
{"constraint", "chk_purchase_spec_resolution_options"},
{"constraint", "chk_purchase_spec_resolution_outcome"},
{"constraint", "chk_purchase_spec_resolution_source"},
{"constraint", "chk_purchase_spec_resolution_confidence"},
{"constraint", "chk_purchase_spec_resolution_decision"},
} {
var exists bool
var err error
if item.kind == "index" {
exists, err = mysqlIndexExists(db, "purchase_spec_resolutions", item.name)
} else {
exists, err = mysqlConstraintExists(db, "purchase_spec_resolutions", item.name)
}
if err != nil || !exists {
return fmt.Errorf("采购运行时规格解析%s %s 缺失: %v", item.kind, item.name, err)
}
}
return nil
}
func checkMySQLV25Shape(db *sql.DB) error {
admin/repository/mysql_db_integration_test.go
+94
View File
@@ -1166,6 +1166,100 @@ func TestMySQLMigrate_V24升级V25且后台队列状态有效(t *testing.T) {
}
}
func TestMySQLMigrate_V25升级V26且规格解析审计幂等(t *testing.T) {
db := openMySQLMigrationTestDB(t)
defer db.Close()
cleanMySQLTestSchema(t, db)
defer cleanMySQLTestSchema(t, db)
if err := MigrateMySQL(db); err != nil {
t.Fatal(err)
}
mustExec(t, db, `DROP TABLE purchase_spec_resolutions`)
mustExec(t, db, `DELETE FROM schema_migrations WHERE version>=26`)
if err := MigrateMySQL(db); err != nil {
t.Fatalf("v25 升级 v26 失败: %v", err)
}
if err := MigrateMySQL(db); err != nil {
t.Fatalf("v26 重复迁移失败: %v", err)
}
if err := checkMySQLV26Shape(db); err != nil {
t.Fatal(err)
}
createdAt := "2026-08-17T08:00:00Z"
resolution := model.PurchaseSpecResolution{
ResolutionID: "PSR-1",
TaskID: "cg254",
AttemptID: "attempt-254",
ClientID: "client-254",
TaskVersion: 3,
PddGoodsID: "937122477375",
OriginalOptionsJSON: `{"color":"黑色","size":"60公斤"}`,
SelectedColor: "黑色",
TargetSize: "60公斤",
CandidatesJSON: `[{"candidate_id":"c1","raw_text":"120斤","options":{"color":"黑色","size":"120斤"}}]`,
CandidateSnapshotHash: strings.Repeat("a", 64),
RequestHash: strings.Repeat("b", 64),
ObservedAt: createdAt,
CreatedAt: createdAt,
}
if err := InsertPurchaseSpecResolution(db, resolution); err != nil {
t.Fatal(err)
}
replayed, err := GetPurchaseSpecResolutionForReplay(db, resolution.TaskID, resolution.AttemptID,
resolution.CandidateSnapshotHash, resolution.RequestHash)
if err != nil || replayed == nil || replayed.ResolutionID != resolution.ResolutionID {
t.Fatalf("相同请求未命中解析记录: resolution=%+v err=%v", replayed, err)
}
if _, err := GetPurchaseSpecResolutionForReplay(db, resolution.TaskID, resolution.AttemptID,
resolution.CandidateSnapshotHash, strings.Repeat("c", 64)); !errors.Is(err, ErrPurchaseSpecResolutionConflict) {
t.Fatalf("相同业务身份的不同请求应冲突，实际 %v", err)
}
duplicate := resolution
duplicate.ResolutionID = "PSR-2"
if err := InsertPurchaseSpecResolution(db, duplicate); !errors.Is(err, ErrPurchaseSpecResolutionExists) {
t.Fatalf("业务身份唯一键未生效，实际 %v", err)
}
decision := model.PurchaseSpecResolutionDecision{
ResolutionID: resolution.ResolutionID,
Outcome: model.PurchaseSpecResolutionMatched,
DecisionSource: model.PurchaseSpecResolutionAI,
ChosenCandidateID: "c1",
ResolvedOptionsJSON: `{"color":"黑色","size":"120斤"}`,
ConfidenceBPS: 9300,
ConfidenceSet: true,
Reason: "候选唯一且通过门禁",
ProviderID: "AI-1",
SourceModel: "model-1",
ConfigFingerprint: strings.Repeat("d", 64),
RulesVersion: "runtime-v1",
PromptVersion: "runtime-v1",
DecidedAt: "2026-08-17T08:00:01Z",
}
if err := CompletePurchaseSpecResolution(db, decision); err != nil {
t.Fatal(err)
}
completed, err := GetPurchaseSpecResolutionByID(db, resolution.ResolutionID)
if err != nil || completed == nil || completed.Outcome != model.PurchaseSpecResolutionMatched ||
completed.DecisionSource != model.PurchaseSpecResolutionAI || !completed.ConfidenceSet || completed.ConfidenceBPS != 9300 {
t.Fatalf("解析决策读取不完整: resolution=%+v err=%v", completed, err)
}
if err := CompletePurchaseSpecResolution(db, decision); !errors.Is(err, ErrPurchaseSpecResolutionCompleted) {
t.Fatalf("已经完成的决策不应被覆盖，实际 %v", err)
}
tooManyCandidates := resolution
tooManyCandidates.ResolutionID = "PSR-TOO-MANY"
tooManyCandidates.AttemptID = "attempt-too-many"
tooManyCandidates.CandidateSnapshotHash = strings.Repeat("e", 64)
tooManyCandidates.RequestHash = strings.Repeat("f", 64)
tooManyCandidates.CandidatesJSON = `[` + strings.TrimSuffix(strings.Repeat(`{"candidate_id":"c1"},`, 101), ",") + `]`
if err := InsertPurchaseSpecResolution(db, tooManyCandidates); err == nil {
t.Fatal("数据库必须拒绝超过 100 个候选的解析观察")
}
}
func TestUpsertInnerCodeImportRow_软删除记录按状态安全恢复(t *testing.T) {
db := openMySQLMigrationTestDB(t)
defer db.Close()
admin/repository/purchase_spec_resolution.go
+143
View File
@@ -0,0 +1,143 @@
package repository
import (
"database/sql"
"errors"
"fmt"
"github.com/go-sql-driver/mysql"
"cmautobuy/admin/model"
)
var (
// ErrPurchaseSpecResolutionExists 表示解析编号或业务身份已经存在，调用方应读取旧记录复查。
ErrPurchaseSpecResolutionExists = errors.New("采购运行时规格解析记录已存在")
// ErrPurchaseSpecResolutionConflict 表示相同业务身份对应了不同请求内容。
ErrPurchaseSpecResolutionConflict = errors.New("采购运行时规格解析请求冲突")
// ErrPurchaseSpecResolutionNotFound 表示要完成的解析记录不存在。
ErrPurchaseSpecResolutionNotFound = errors.New("采购运行时规格解析记录不存在")
// ErrPurchaseSpecResolutionCompleted 表示解析记录已经完成，不能覆盖第一次决策。
ErrPurchaseSpecResolutionCompleted = errors.New("采购运行时规格解析记录已经完成")
)
// InsertPurchaseSpecResolution 保存一次候选观察。匹配逻辑不在 Repository 中执行。
func InsertPurchaseSpecResolution(q Execer, resolution model.PurchaseSpecResolution) error {
if resolution.Outcome == "" {
resolution.Outcome = model.PurchaseSpecResolutionPending
}
_, err := q.Exec(`INSERT INTO purchase_spec_resolutions
(resolution_id,task_id,attempt_id,client_id,task_version,pdd_goods_id,
original_options_json,selected_color,target_size,candidates_json,candidate_snapshot_hash,
request_hash,observed_at,outcome,created_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
resolution.ResolutionID, resolution.TaskID, resolution.AttemptID, resolution.ClientID,
resolution.TaskVersion, resolution.PddGoodsID, resolution.OriginalOptionsJSON,
resolution.SelectedColor, resolution.TargetSize, resolution.CandidatesJSON,
resolution.CandidateSnapshotHash, resolution.RequestHash, resolution.ObservedAt,
resolution.Outcome, resolution.CreatedAt)
if err == nil {
return nil
}
var mysqlErr *mysql.MySQLError
if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 {
return ErrPurchaseSpecResolutionExists
}
return fmt.Errorf("写入采购运行时规格解析观察失败: %w", err)
}
// GetPurchaseSpecResolutionByID 按公开解析编号读取候选观察和最终决策。
func GetPurchaseSpecResolutionByID(q Execer, resolutionID string) (*model.PurchaseSpecResolution, error) {
return scanPurchaseSpecResolution(q.QueryRow(purchaseSpecResolutionSelect+` WHERE resolution_id=?`, resolutionID))
}
// GetPurchaseSpecResolutionByIdentity 按任务、执行尝试和候选快照读取唯一解析记录。
func GetPurchaseSpecResolutionByIdentity(q Execer, taskID, attemptID, candidateSnapshotHash string) (*model.PurchaseSpecResolution, error) {
return scanPurchaseSpecResolution(q.QueryRow(purchaseSpecResolutionSelect+
` WHERE task_id=? AND attempt_id=? AND candidate_snapshot_hash=?`,
taskID, attemptID, candidateSnapshotHash))
}
// GetPurchaseSpecResolutionForReplay 复查幂等重放。相同业务身份但请求哈希不同必须冲突。
func GetPurchaseSpecResolutionForReplay(q Execer, taskID, attemptID, candidateSnapshotHash, requestHash string) (*model.PurchaseSpecResolution, error) {
resolution, err := GetPurchaseSpecResolutionByIdentity(q, taskID, attemptID, candidateSnapshotHash)
if err != nil || resolution == nil {
return resolution, err
}
if resolution.RequestHash != requestHash {
return nil, ErrPurchaseSpecResolutionConflict
}
return resolution, nil
}
// CompletePurchaseSpecResolution 只允许把 pending 记录完成一次，保留第一次决策审计。
func CompletePurchaseSpecResolution(q Execer, decision model.PurchaseSpecResolutionDecision) error {
result, err := q.Exec(`UPDATE purchase_spec_resolutions SET
outcome=?,decision_source=?,chosen_candidate_id=?,resolved_options_json=?,confidence_bps=?,
reason=?,provider_id=?,source_model=?,config_fingerprint=?,rules_version=?,prompt_version=?,decided_at=?
WHERE resolution_id=? AND outcome='pending'`,
decision.Outcome, nullableText(string(decision.DecisionSource)), nullableText(decision.ChosenCandidateID),
nullableText(decision.ResolvedOptionsJSON), nullableInt(decision.ConfidenceBPS, decision.ConfidenceSet),
nullableText(decision.Reason), nullableText(decision.ProviderID), nullableText(decision.SourceModel),
nullableText(decision.ConfigFingerprint), nullableText(decision.RulesVersion), nullableText(decision.PromptVersion),
decision.DecidedAt, decision.ResolutionID)
if err != nil {
return fmt.Errorf("完成采购运行时规格解析失败: %w", err)
}
affected, err := result.RowsAffected()
if err != nil {
return fmt.Errorf("读取采购运行时规格解析更新结果失败: %w", err)
}
if affected == 1 {
return nil
}
existing, err := GetPurchaseSpecResolutionByID(q, decision.ResolutionID)
if err != nil {
return err
}
if existing == nil {
return ErrPurchaseSpecResolutionNotFound
}
return ErrPurchaseSpecResolutionCompleted
}
const purchaseSpecResolutionSelect = `SELECT
resolution_id,task_id,attempt_id,client_id,task_version,pdd_goods_id,
original_options_json,selected_color,target_size,candidates_json,candidate_snapshot_hash,
request_hash,observed_at,outcome,decision_source,chosen_candidate_id,resolved_options_json,
confidence_bps,reason,provider_id,source_model,config_fingerprint,rules_version,prompt_version,
created_at,decided_at FROM purchase_spec_resolutions`
func scanPurchaseSpecResolution(row *sql.Row) (*model.PurchaseSpecResolution, error) {
var resolution model.PurchaseSpecResolution
var source, candidateID, resolvedOptions, reason, providerID sql.NullString
var sourceModel, configFingerprint, rulesVersion, promptVersion, decidedAt sql.NullString
var confidence sql.NullInt64
err := row.Scan(
&resolution.ResolutionID, &resolution.TaskID, &resolution.AttemptID, &resolution.ClientID,
&resolution.TaskVersion, &resolution.PddGoodsID, &resolution.OriginalOptionsJSON,
&resolution.SelectedColor, &resolution.TargetSize, &resolution.CandidatesJSON,
&resolution.CandidateSnapshotHash, &resolution.RequestHash, &resolution.ObservedAt,
&resolution.Outcome, &source, &candidateID, &resolvedOptions, &confidence, &reason,
&providerID, &sourceModel, &configFingerprint, &rulesVersion, &promptVersion,
&resolution.CreatedAt, &decidedAt)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("读取采购运行时规格解析失败: %w", err)
}
resolution.DecisionSource = model.PurchaseSpecResolutionSource(source.String)
resolution.ChosenCandidateID = candidateID.String
resolution.ResolvedOptionsJSON = resolvedOptions.String
resolution.ConfidenceBPS = int(confidence.Int64)
resolution.ConfidenceSet = confidence.Valid
resolution.Reason = reason.String
resolution.ProviderID = providerID.String
resolution.SourceModel = sourceModel.String
resolution.ConfigFingerprint = configFingerprint.String
resolution.RulesVersion = rulesVersion.String
resolution.PromptVersion = promptVersion.String
resolution.DecidedAt = decidedAt.String
return &resolution, nil
}