feat: 实现采购运行时规格解析 (#255)

This commit is contained in:
chengma
2026-08-17 17:20:07 +08:00
parent a699ec8132
commit de36cfd0b5
10 changed files with 1347 additions and 42 deletions
+64 -5
View File
@@ -3,7 +3,7 @@
// **权威契约在 Client 那边**:docs/client/04-admin-api-contract.md。
// 本包只是实现,两边说法不一致时以 Client 契约为准,要改先走工单。
//
// 一共只有三个接口,**不得新增"让 Client 查询状态"类接口**,
// 接口都是一次性命令,**不得新增"让 Client 查询状态"类接口**,
// 也不得加回租约和心跳,理由见 docs/admin/04-client-api.md §1。
package api
@@ -23,12 +23,14 @@ import (
// Handler 持有接口共用的依赖。
type Handler struct {
db *sql.DB
db *sql.DB
aiSecrets service.AISecretStore
aiPolicy service.AIEndpointPolicy
}
// Register 挂上三个接口。
func Register(r *gin.Engine, db *sql.DB) {
h := &Handler{db: db}
// Register 挂上 Client 命令接口。
func Register(r *gin.Engine, db *sql.DB, aiSecrets service.AISecretStore, aiPolicy service.AIEndpointPolicy) {
h := &Handler{db: db, aiSecrets: aiSecrets, aiPolicy: aiPolicy}
g := r.Group("/api/v1/client")
{
@@ -38,6 +40,7 @@ func Register(r *gin.Engine, db *sql.DB) {
g.POST("/tasks/claim", h.Claim)
g.POST("/tasks/:task_id/result", h.SubmitResult)
g.POST("/tasks/:task_id/failure", h.SubmitFailure)
g.POST("/tasks/:task_id/spec-resolution", h.ResolvePurchaseSpec)
}
}
@@ -203,6 +206,62 @@ func (h *Handler) SubmitFailure(c *gin.Context) {
h.handleSubmit(c, service.SubmitFailure, "task_failure_received")
}
// ResolvePurchaseSpec 接收 Client 当前 PDD 页面上的可购买尺码快照。
// 这是一次性决策命令,不查询或修改任务状态,也不会绕过 Client 的下单复核门禁。
func (h *Handler) ResolvePurchaseSpec(c *gin.Context) {
taskID := c.Param("task_id")
clientID := c.GetHeader("X-Client-Id")
if clientID == "" {
apiError(c, http.StatusBadRequest, "MISSING_CLIENT_ID", "缺少 X-Client-Id 请求头", false)
return
}
idempotencyKey := c.GetHeader("Idempotency-Key")
if idempotencyKey == "" {
apiError(c, http.StatusBadRequest, "MISSING_IDEMPOTENCY_KEY", "缺少 Idempotency-Key 请求头", false)
return
}
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, service.MaxPurchaseSpecResolutionBodyBytes)
rawBody, err := io.ReadAll(c.Request.Body)
if err != nil {
apiError(c, http.StatusBadRequest, "INVALID_BODY", "请求体不是合法 JSON 或超过 64 KiB", false)
return
}
responseJSON, err := service.ResolvePurchaseSpec(c.Request.Context(), h.db, h.aiSecrets, h.aiPolicy,
taskID, clientID, idempotencyKey, rawBody)
switch {
case err == nil:
log.Printf("purchase_spec_resolved client_id=%s task_id=%s", clientID, taskID)
c.Data(http.StatusOK, "application/json; charset=utf-8", []byte(responseJSON))
case errors.Is(err, service.ErrInvalidSpecResolutionBody):
apiError(c, http.StatusBadRequest, "INVALID_BODY", "请求体不是合法 JSON 或超过 64 KiB", false)
case errors.Is(err, service.ErrInvalidSpecResolutionSchema):
apiError(c, http.StatusBadRequest, "INVALID_SPEC_RESOLUTION_SCHEMA", "不支持的规格解析 schema_version", false)
case errors.Is(err, service.ErrInvalidSpecResolutionRequest):
apiError(c, http.StatusUnprocessableEntity, "INVALID_SPEC_RESOLUTION_REQUEST", "规格解析字段或任务目标规格无效", false)
case errors.Is(err, service.ErrSpecResolutionHashMismatch):
apiError(c, http.StatusUnprocessableEntity, "SPEC_RESOLUTION_HASH_MISMATCH", "候选快照哈希或幂等键不一致", false)
case errors.Is(err, service.ErrTaskNotFound):
apiError(c, http.StatusNotFound, "TASK_NOT_FOUND", "任务不存在", false)
case errors.Is(err, service.ErrTaskNotPurchase):
apiError(c, http.StatusUnprocessableEntity, "TASK_NOT_PURCHASE", "任务不是采购任务", false)
case errors.Is(err, service.ErrTaskVersionConflict):
apiError(c, http.StatusConflict, "TASK_VERSION_CONFLICT", "任务版本已变化", false)
case errors.Is(err, service.ErrPDDGoodsMismatch):
apiError(c, http.StatusUnprocessableEntity, "PDD_GOODS_MISMATCH", "PDD 商品与任务不一致", false)
case errors.Is(err, service.ErrNeverClaimed):
apiError(c, http.StatusForbidden, "TASK_NOT_CLAIMED_BY_CLIENT", "该 Client 从未领取过此任务", false)
case errors.Is(err, service.ErrIdempotencyConflict):
apiError(c, http.StatusConflict, "IDEMPOTENCY_CONFLICT", "相同幂等身份提交了不同内容", false)
case errors.Is(err, service.ErrSpecResolutionUnavailable):
log.Printf("purchase_spec_unavailable client_id=%s task_id=%s err=%v", clientID, taskID, err)
apiError(c, http.StatusServiceUnavailable, "SPEC_RESOLUTION_UNAVAILABLE", "规格解析服务暂不可用,请有限重试原请求", true)
default:
log.Printf("purchase_spec_failed client_id=%s task_id=%s err=%v", clientID, taskID, err)
apiError(c, http.StatusInternalServerError, "SPEC_RESOLUTION_FAILED", "规格解析失败,请稍后重试", true)
}
}
// submitFunc 是两个提交接口共用的处理函数形状。
type submitFunc func(db *sql.DB, taskID, clientID, idemKey string, rawBody []byte) (string, error)
+24
View File
@@ -1,9 +1,16 @@
package api
import (
"bytes"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
"cmautobuy/admin/model"
"cmautobuy/admin/service"
)
func TestTaskPayload_包含不可变执行模式(t *testing.T) {
@@ -12,3 +19,20 @@ func TestTaskPayload_包含不可变执行模式(t *testing.T) {
t.Fatalf("execution_mode=%v", payload["execution_mode"])
}
}
func TestResolvePurchaseSpec_请求体上限在进入数据库前拒绝(t *testing.T) {
gin.SetMode(gin.TestMode)
router := gin.New()
Register(router, nil, nil, service.AIEndpointPolicy{})
body := bytes.Repeat([]byte("x"), service.MaxPurchaseSpecResolutionBodyBytes+1)
request := httptest.NewRequest(http.MethodPost,
"/api/v1/client/tasks/cg255/spec-resolution", bytes.NewReader(body))
request.Header.Set("X-Client-Id", "CLIENT-255")
request.Header.Set("Idempotency-Key", "spec-resolution-v1:"+strings.Repeat("0", 64))
recorder := httptest.NewRecorder()
router.ServeHTTP(recorder, request)
if recorder.Code != http.StatusBadRequest || !strings.Contains(recorder.Body.String(), `"code":"INVALID_BODY"`) {
t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
}
}
+1 -1
View File
@@ -128,7 +128,7 @@ func newRouter(db *sql.DB) (*gin.Engine, error) {
aiSecrets := service.NewFileAISecretStore(aiConfig.SecretsPath)
aiPolicy := service.NewAIEndpointPolicy(aiConfig.AllowedHosts)
web.Register(r, db, config.OnlineThreshold, aiSecrets, aiPolicy) // 给浏览器的页面
api.Register(r, db) // 给 Client 的接口
api.Register(r, db, aiSecrets, aiPolicy) // 给 Client 的接口
catalogConfig, err := config.LoadCatalogIntegration()
if err != nil {
return nil, fmt.Errorf("读取商品目录接口配置失败: %w", err)
+10 -4
View File
@@ -306,20 +306,25 @@ func HasEverClaimed(q Execer, taskID, clientID string) (bool, error) {
// TaskInfo 是提交结果时需要知道的任务基本信息。
type TaskInfo struct {
TaskType model.TaskType
// Version 是 Client 领取任务时拿到的乐观版本号。
Version int64
// GoodsID 是关联的**蝦皮**商品编号。
GoodsID string
// PddGoodsID 是要采集/购买的**拼多多**商品编号。
// 采集结果落到 pdd_products 时用的是它,不是 GoodsID —— 被采集的是 PDD 商品。
PddGoodsID string
// PddOptions 是任务创建时固定的目标规格 JSON。
// 运行时规格解析必须逐项核对,不能接受 Client 临时改写目标规格。
PddOptions string
}
// GetTaskInfo 查任务的类型和关联商品。任务不存在时返回 (nil, nil)。
// GetTaskInfo 查任务的类型、版本和关联商品。任务不存在时返回 (nil, nil)。
func GetTaskInfo(q Execer, taskID string) (*TaskInfo, error) {
var info TaskInfo
var gid, pddGID sql.NullString
var gid, pddGID, pddOptions sql.NullString
err := q.QueryRow(
`SELECT task_type, goods_id, pdd_goods_id FROM tasks WHERE task_id = ?`,
taskID).Scan(&info.TaskType, &gid, &pddGID)
`SELECT task_type,version,goods_id,pdd_goods_id,pdd_options FROM tasks WHERE task_id = ?`,
taskID).Scan(&info.TaskType, &info.Version, &gid, &pddGID, &pddOptions)
if err == sql.ErrNoRows {
return nil, nil
}
@@ -328,6 +333,7 @@ func GetTaskInfo(q Execer, taskID string) (*TaskInfo, error) {
}
info.GoodsID = gid.String
info.PddGoodsID = pddGID.String
info.PddOptions = pddOptions.String
return &info, nil
}
+16 -32
View File
@@ -153,55 +153,39 @@ func MatchSybSpecWithAI(ctx context.Context, db *sql.DB, actor *model.User, snap
}
bindings := bindAICandidates(eligible)
baseDecision.CandidatesJSON = candidateAuditJSON(bindings)
if snapshot.Client == nil || snapshot.Secret == "" || snapshot.Provider.ProviderID == "" {
return recordAIMatchWithoutSave(db, baseDecision, "failed", "AI 服务商运行快照不可用")
}
request := AIModelMatchRequest{ProductTitle: truncateRunes(orderContext.Order.Title, 300), SourceSpec: truncateRunes(orderContext.Order.ProductSpec, 300)}
for _, binding := range bindings {
request.Candidates = append(request.Candidates, AIModelCandidate{ID: binding.ID,
Label: truncateRunes(binding.Choice.Label, 300), Options: boundedAIOptions(binding.Choice.Options)})
}
response, modelErr := snapshot.Client.Match(ctx, snapshot.Provider, snapshot.Secret, request)
if modelErr != nil {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "failed", safeAIError(modelErr))
result.ModelCalled = true
return result, auditErr
}
if err := validateAIModelMatchResponse(response); err != nil {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "failed", err.Error())
result.ModelCalled = true
return result, auditErr
}
selection := selectAIWhitelistedCandidate(ctx, snapshot, request)
response := selection.Response
baseDecision.ChosenCandidateID = response.CandidateID
baseDecision.ConfidenceBPS = response.ConfidenceBPS
baseDecision.ConfidenceSet = true
baseDecision.Reason = response.Reason
baseDecision.ConfidenceSet = selection.ConfidenceSet
baseDecision.Reason = selection.Reason
baseDecision.ConflictDimensionsJSON = stringArrayJSON(response.ConflictDimensions)
baseDecision.MissingDimensionsJSON = stringArrayJSON(response.MissingDimensions)
if response.Conclusion != "match" {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "rejected", response.Reason)
result.ModelCalled = true
if selection.Status != aiSelectionMatched {
outcome := "rejected"
if selection.Status == aiSelectionFailed {
outcome = "failed"
}
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, outcome, selection.Reason)
result.ModelCalled = selection.ModelCalled
return result, auditErr
}
selected, found := bindingByID(bindings, response.CandidateID)
if !found {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "rejected", "模型返回了不在候选白名单中的编号")
result.ModelCalled = true
// selectAIWhitelistedCandidate 已核对过同一 request;此分支只防以后
// 绑定构造被改坏时把错误候选写进真实映射。
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "rejected", "候选绑定已变化,未保存映射")
result.ModelCalled = selection.ModelCalled
return result, auditErr
}
baseDecision.ChosenOptionKey = selected.Choice.Key
if len(response.ConflictDimensions) > 0 || len(response.MissingDimensions) > 0 {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "rejected", "模型报告仍有冲突或缺失维度")
result.ModelCalled = true
return result, auditErr
}
if response.ConfidenceBPS < snapshot.Provider.ConfidenceThresholdBPS {
result, auditErr := recordAIMatchWithoutSave(db, baseDecision, "rejected", "模型置信度低于管理员设置的自动写入阈值")
result.ModelCalled = true
return result, auditErr
}
result, err := saveAutomaticMapping(db, *actor, *orderContext, selected.Choice, "ai", baseDecision)
result.ModelCalled = true
result.ModelCalled = selection.ModelCalled
return result, err
}
+72
View File
@@ -39,6 +39,78 @@ type AIModelClient interface {
Match(context.Context, model.AIProviderConfig, string, AIModelMatchRequest) (AIModelMatchResponse, error)
}
type aiWhitelistedSelectionStatus string
const (
aiSelectionMatched aiWhitelistedSelectionStatus = "matched"
aiSelectionUncertain aiWhitelistedSelectionStatus = "uncertain"
aiSelectionConflict aiWhitelistedSelectionStatus = "conflict"
aiSelectionFailed aiWhitelistedSelectionStatus = "failed"
aiSelectionCandidateRejected aiWhitelistedSelectionStatus = "candidate_rejected"
aiSelectionDimensionsUnresolved aiWhitelistedSelectionStatus = "dimensions_unresolved"
aiSelectionBelowThreshold aiWhitelistedSelectionStatus = "below_threshold"
)
// aiWhitelistedSelection 是 SYB 批量匹配和 Client 真机匹配共用的模型安全结论。
// 业务编排可以把“不确定”显示成不同文案,但不能绕过这里的候选、阈值和维度门禁。
type aiWhitelistedSelection struct {
Status aiWhitelistedSelectionStatus
Response AIModelMatchResponse
Reason string
ConfidenceSet bool
ModelCalled bool
}
func selectAIWhitelistedCandidate(ctx context.Context, snapshot AIMatchSnapshot,
request AIModelMatchRequest) aiWhitelistedSelection {
if snapshot.Client == nil || snapshot.Secret == "" || snapshot.Provider.ProviderID == "" {
return aiWhitelistedSelection{Status: aiSelectionFailed, Reason: "AI 服务商运行快照不可用"}
}
response, err := snapshot.Client.Match(ctx, snapshot.Provider, snapshot.Secret, request)
if err != nil {
reason := safeAIError(err)
if snapshot.Secret != "" {
reason = strings.ReplaceAll(reason, snapshot.Secret, "[REDACTED]")
}
return aiWhitelistedSelection{Status: aiSelectionFailed, Reason: reason, ModelCalled: true}
}
result := aiWhitelistedSelection{Response: response, Reason: response.Reason,
ConfidenceSet: true, ModelCalled: true}
if err := validateAIModelMatchResponse(response); err != nil {
result.Status, result.Reason, result.ConfidenceSet = aiSelectionFailed, err.Error(), false
return result
}
if response.Conclusion == "uncertain" {
result.Status = aiSelectionUncertain
return result
}
if response.Conclusion == "conflict" {
result.Status = aiSelectionConflict
return result
}
allowed := false
for _, candidate := range request.Candidates {
if candidate.ID == response.CandidateID {
allowed = true
break
}
}
if !allowed {
result.Status, result.Reason = aiSelectionCandidateRejected, "模型返回了不在候选白名单中的编号"
return result
}
if len(response.ConflictDimensions) > 0 || len(response.MissingDimensions) > 0 {
result.Status, result.Reason = aiSelectionDimensionsUnresolved, "模型报告仍有冲突或缺失维度"
return result
}
if response.ConfidenceBPS < snapshot.Provider.ConfidenceThresholdBPS {
result.Status, result.Reason = aiSelectionBelowThreshold, "模型置信度低于管理员设置的自动选择阈值"
return result
}
result.Status = aiSelectionMatched
return result
}
type OpenAICompatibleModelClient struct{ doer AIHTTPDoer }
func NewOpenAICompatibleModelClient(doer AIHTTPDoer) *OpenAICompatibleModelClient {
+677
View File
@@ -0,0 +1,677 @@
package service
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"reflect"
"regexp"
"strconv"
"strings"
"sync"
"time"
"unicode"
"unicode/utf8"
"cmautobuy/admin/model"
"cmautobuy/admin/repository"
)
const (
// MaxPurchaseSpecResolutionBodyBytes 是 Client 运行时规格解析请求的硬上限。
MaxPurchaseSpecResolutionBodyBytes = 64 << 10
PurchaseSpecRulesVersion = "purchase_rules_v1"
purchaseSpecSchemaVersion = 1
)
var (
ErrInvalidSpecResolutionBody = errors.New("采购运行时规格解析请求体无效")
ErrInvalidSpecResolutionSchema = errors.New("采购运行时规格解析 schema 不支持")
ErrInvalidSpecResolutionRequest = errors.New("采购运行时规格解析字段无效")
ErrSpecResolutionHashMismatch = errors.New("采购运行时规格解析哈希不一致")
ErrTaskNotPurchase = errors.New("任务不是采购任务")
ErrTaskVersionConflict = errors.New("任务版本不一致")
ErrPDDGoodsMismatch = errors.New("PDD 商品不一致")
ErrSpecResolutionUnavailable = errors.New("采购运行时规格解析服务暂不可用")
)
// PurchaseSpecCandidate 是 Client 当前页面上一个可购买尺码的逐字快照。
// candidate_id 只在本次请求内有效,AI 只能从这些短编号中选择。
type PurchaseSpecCandidate struct {
CandidateID string `json:"candidate_id"`
RawText string `json:"raw_text"`
Options map[string]string `json:"options"`
}
// PurchaseSpecResolutionRequest 是 docs/client/04-admin-api-contract.md §7.1 的请求体。
type PurchaseSpecResolutionRequest struct {
SchemaVersion int `json:"schema_version"`
TaskVersion int64 `json:"task_version"`
AttemptID string `json:"attempt_id"`
PddGoodsID string `json:"pdd_goods_id"`
OriginalOptions map[string]string `json:"original_options"`
SelectedColor string `json:"selected_color"`
TargetSize string `json:"target_size"`
Candidates []PurchaseSpecCandidate `json:"candidates"`
CandidateSnapshotHash string `json:"candidate_snapshot_hash"`
ObservedAt string `json:"observed_at"`
}
type PurchaseSpecResolutionMatch struct {
CandidateID string `json:"candidate_id"`
RawText string `json:"raw_text"`
Options map[string]string `json:"options"`
}
// PurchaseSpecResolutionResponse 只返回请求候选的逐字副本,不返回模型生成的新规格。
type PurchaseSpecResolutionResponse struct {
SchemaVersion int `json:"schema_version"`
ResolutionID string `json:"resolution_id"`
Outcome model.PurchaseSpecResolutionOutcome `json:"outcome"`
Source *model.PurchaseSpecResolutionSource `json:"source"`
CandidateSnapshotHash string `json:"candidate_snapshot_hash"`
Match *PurchaseSpecResolutionMatch `json:"match"`
ConfidenceBPS *int `json:"confidence_bps"`
Reason string `json:"reason"`
ResolvedAt string `json:"resolved_at"`
}
type purchaseSpecDecision struct {
Outcome model.PurchaseSpecResolutionOutcome
Source model.PurchaseSpecResolutionSource
Candidate *PurchaseSpecCandidate
ConfidenceBPS int
ConfidenceSet bool
Reason string
ProviderID string
SourceModel string
Fingerprint string
PromptVersion string
}
type purchaseSpecGate struct {
token chan struct{}
refs int
}
var purchaseSpecGates = struct {
sync.Mutex
items map[string]*purchaseSpecGate
}{items: map[string]*purchaseSpecGate{}}
// ResolvePurchaseSpec 保存一次真机候选观察,先跑确定性规则,确实不能唯一判断时才调用 AI。
// 它只写解析审计和幂等响应,不修改任务、PDD 主数据或任何下单安全字段。
func ResolvePurchaseSpec(ctx context.Context, db *sql.DB, secrets AISecretStore, policy AIEndpointPolicy,
taskID, clientID, idempotencyKey string, rawBody []byte) (string, error) {
return resolvePurchaseSpecWithSnapshotLoader(ctx, db, secrets, policy, taskID, clientID,
idempotencyKey, rawBody, LoadActiveAIMatchSnapshot)
}
type aiSnapshotLoader func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error)
func resolvePurchaseSpecWithSnapshotLoader(ctx context.Context, db *sql.DB, secrets AISecretStore,
policy AIEndpointPolicy, taskID, clientID, idempotencyKey string, rawBody []byte,
loadSnapshot aiSnapshotLoader) (string, error) {
req, err := parsePurchaseSpecResolutionRequest(taskID, idempotencyKey, rawBody)
if err != nil {
return "", err
}
requestHash := repository.HashRequest(rawBody)
// 同一 Admin 进程内,同一个业务身份只允许一个协程进入模型调用。
// 锁不包含数据库事务;不同任务仍可以并行解析。
release, err := acquirePurchaseSpecGate(ctx, idempotencyKey)
if err != nil {
return "", fmt.Errorf("%w: %v", ErrSpecResolutionUnavailable, err)
}
defer release()
resolution, leader, replay, err := beginPurchaseSpecResolution(
db, taskID, clientID, idempotencyKey, requestHash, req)
if err != nil || replay != "" {
return replay, err
}
if !leader && resolution.Outcome != model.PurchaseSpecResolutionPending {
return saveReconstructedPurchaseSpecResponse(db, idempotencyKey, requestHash, resolution, req.Candidates)
}
// pending 可能来自进程中断后的原请求。因为本进程已取得该业务身份的唯一门禁,
// 可以安全接管;最终 UPDATE 仍只允许 pending 完成一次。
decision, terminal := matchPurchaseSpecByRule(req.TargetSize, req.Candidates)
if !terminal {
decision = matchPurchaseSpecWithAI(ctx, db, secrets, policy, req, loadSnapshot)
}
if purchaseSpecDiagnosticSource(db, req.PddGoodsID) == "shopee_backfill" {
decision.Reason = truncateRunes(decision.Reason+";诊断来源:shopee_backfill", 500)
}
return completePurchaseSpecResolution(db, idempotencyKey, requestHash, resolution, req.Candidates, decision)
}
func parsePurchaseSpecResolutionRequest(taskID, idempotencyKey string, rawBody []byte) (PurchaseSpecResolutionRequest, error) {
var req PurchaseSpecResolutionRequest
if len(rawBody) == 0 || len(rawBody) > MaxPurchaseSpecResolutionBodyBytes || !utf8.Valid(rawBody) {
return req, ErrInvalidSpecResolutionBody
}
if err := json.Unmarshal(rawBody, &req); err != nil {
return req, ErrInvalidSpecResolutionBody
}
if req.SchemaVersion != purchaseSpecSchemaVersion {
return req, ErrInvalidSpecResolutionSchema
}
if req.TaskVersion <= 0 || !validSpecText(req.AttemptID) || !validSpecText(req.PddGoodsID) ||
!validSpecText(req.SelectedColor) || !validSpecText(req.TargetSize) ||
len(req.OriginalOptions) < 1 || len(req.OriginalOptions) > 16 ||
len(req.Candidates) < 1 || len(req.Candidates) > 100 {
return req, ErrInvalidSpecResolutionRequest
}
for key, value := range req.OriginalOptions {
if !validSpecText(key) || !validSpecText(value) {
return req, ErrInvalidSpecResolutionRequest
}
}
for index, candidate := range req.Candidates {
if candidate.CandidateID != "c"+strconv.Itoa(index+1) || !validSpecText(candidate.RawText) ||
len(candidate.Options) != 2 || candidate.Options["color"] != req.SelectedColor ||
candidate.Options["size"] != candidate.RawText {
return req, ErrInvalidSpecResolutionRequest
}
if _, ok := candidate.Options["color"]; !ok {
return req, ErrInvalidSpecResolutionRequest
}
if _, ok := candidate.Options["size"]; !ok {
return req, ErrInvalidSpecResolutionRequest
}
}
if _, err := time.Parse(time.RFC3339Nano, req.ObservedAt); err != nil {
return req, ErrInvalidSpecResolutionRequest
}
expectedSnapshot := purchaseSpecCandidateSnapshotHash(req)
if req.CandidateSnapshotHash != expectedSnapshot || idempotencyKey != purchaseSpecIdempotencyKey(taskID, req) {
return req, ErrSpecResolutionHashMismatch
}
return req, nil
}
func validSpecText(value string) bool {
if value == "" || utf8.RuneCountInString(value) > 191 {
return false
}
for _, r := range value {
if unicode.IsControl(r) {
return false
}
}
return true
}
func purchaseSpecCandidateSnapshotHash(req PurchaseSpecResolutionRequest) string {
var material strings.Builder
material.WriteString(specFrame("spec-resolution-v1"))
material.WriteString(specFrame(req.PddGoodsID))
material.WriteString(specFrame(req.SelectedColor))
material.WriteString(specFrame(strconv.Itoa(len(req.Candidates))))
for _, candidate := range req.Candidates {
material.WriteString(specFrame(candidate.CandidateID))
material.WriteString(specFrame(candidate.RawText))
material.WriteString(specFrame(candidate.Options["color"]))
material.WriteString(specFrame(candidate.Options["size"]))
}
return sha256Hex(material.String())
}
func purchaseSpecIdempotencyKey(taskID string, req PurchaseSpecResolutionRequest) string {
material := specFrame(taskID) + specFrame(req.AttemptID) +
specFrame(req.CandidateSnapshotHash) + specFrame("spec-resolution-v1")
return "spec-resolution-v1:" + sha256Hex(material)
}
func specFrame(value string) string { return strconv.Itoa(len([]byte(value))) + ":" + value }
func sha256Hex(value string) string {
sum := sha256.Sum256([]byte(value))
return hex.EncodeToString(sum[:])
}
func acquirePurchaseSpecGate(ctx context.Context, key string) (func(), error) {
purchaseSpecGates.Lock()
gate := purchaseSpecGates.items[key]
if gate == nil {
gate = &purchaseSpecGate{token: make(chan struct{}, 1)}
purchaseSpecGates.items[key] = gate
}
gate.refs++
purchaseSpecGates.Unlock()
select {
case gate.token <- struct{}{}:
return func() {
<-gate.token
purchaseSpecGates.Lock()
gate.refs--
if gate.refs == 0 {
delete(purchaseSpecGates.items, key)
}
purchaseSpecGates.Unlock()
}, nil
case <-ctx.Done():
purchaseSpecGates.Lock()
gate.refs--
if gate.refs == 0 {
delete(purchaseSpecGates.items, key)
}
purchaseSpecGates.Unlock()
return nil, ctx.Err()
}
}
func beginPurchaseSpecResolution(db *sql.DB, taskID, clientID, idempotencyKey, requestHash string,
req PurchaseSpecResolutionRequest) (*model.PurchaseSpecResolution, bool, string, error) {
tx, err := db.Begin()
if err != nil {
return nil, false, "", fmt.Errorf("%w: 开始候选观察事务失败", ErrSpecResolutionUnavailable)
}
defer tx.Rollback()
if body, done, lookupErr := repository.LookupIdempotent(tx, idempotencyKey, requestHash); lookupErr != nil {
if errors.Is(lookupErr, repository.ErrIdempotencyConflict) {
return nil, false, "", ErrIdempotencyConflict
}
return nil, false, "", fmt.Errorf("%w: 查询幂等结果失败", ErrSpecResolutionUnavailable)
} else if done {
return nil, false, body, nil
}
info, err := repository.GetTaskInfo(tx, taskID)
if err != nil {
return nil, false, "", fmt.Errorf("%w: 查询任务失败", ErrSpecResolutionUnavailable)
}
if info == nil {
return nil, false, "", ErrTaskNotFound
}
if info.TaskType != model.TaskPurchase {
return nil, false, "", ErrTaskNotPurchase
}
if info.Version != req.TaskVersion {
return nil, false, "", ErrTaskVersionConflict
}
if info.PddGoodsID != req.PddGoodsID {
return nil, false, "", ErrPDDGoodsMismatch
}
if !taskOptionsEqual(info.PddOptions, req.OriginalOptions) ||
!targetOptionWasClaimed(req.OriginalOptions, "color", req.SelectedColor) ||
!targetOptionWasClaimed(req.OriginalOptions, "size", req.TargetSize) {
return nil, false, "", ErrInvalidSpecResolutionRequest
}
claimed, err := repository.HasEverClaimed(tx, taskID, clientID)
if err != nil {
return nil, false, "", fmt.Errorf("%w: 查询任务领取历史失败", ErrSpecResolutionUnavailable)
}
if !claimed {
return nil, false, "", ErrNeverClaimed
}
originalJSON, _ := json.Marshal(req.OriginalOptions)
candidatesJSON, _ := json.Marshal(req.Candidates)
resolution := &model.PurchaseSpecResolution{
ResolutionID: "psr-" + newID(), TaskID: taskID, AttemptID: req.AttemptID, ClientID: clientID,
TaskVersion: req.TaskVersion, PddGoodsID: req.PddGoodsID,
OriginalOptionsJSON: string(originalJSON), SelectedColor: req.SelectedColor, TargetSize: req.TargetSize,
CandidatesJSON: string(candidatesJSON), CandidateSnapshotHash: req.CandidateSnapshotHash,
RequestHash: requestHash, ObservedAt: req.ObservedAt, Outcome: model.PurchaseSpecResolutionPending,
CreatedAt: model.NowISO(),
}
if err := repository.InsertPurchaseSpecResolution(tx, *resolution); err != nil {
if !errors.Is(err, repository.ErrPurchaseSpecResolutionExists) {
return nil, false, "", fmt.Errorf("%w: 保存候选观察失败", ErrSpecResolutionUnavailable)
}
// MySQL 默认 REPEATABLE READ:本事务前面已经做过一致性读,唯一键等待
// 另一事务提交后,继续在原事务查询仍可能看不到赢家。先回滚再用新快照复查。
if rollbackErr := tx.Rollback(); rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
return nil, false, "", fmt.Errorf("%w: 回滚重复候选观察失败", ErrSpecResolutionUnavailable)
}
existing, replayErr := repository.GetPurchaseSpecResolutionForReplay(
db, taskID, req.AttemptID, req.CandidateSnapshotHash, requestHash)
if errors.Is(replayErr, repository.ErrPurchaseSpecResolutionConflict) {
return nil, false, "", ErrIdempotencyConflict
}
if replayErr != nil || existing == nil {
return nil, false, "", fmt.Errorf("%w: 读取已有候选观察失败", ErrSpecResolutionUnavailable)
}
return existing, false, "", nil
}
if err := tx.Commit(); err != nil {
return nil, false, "", fmt.Errorf("%w: 提交候选观察失败", ErrSpecResolutionUnavailable)
}
return resolution, true, "", nil
}
func taskOptionsEqual(raw string, submitted map[string]string) bool {
var expected map[string]string
return json.Unmarshal([]byte(raw), &expected) == nil && reflect.DeepEqual(expected, submitted)
}
func targetOptionWasClaimed(options map[string]string, preferredKey, target string) bool {
if value, ok := options[preferredKey]; ok {
return value == target
}
for _, value := range options {
if value == target {
return true
}
}
return false
}
func purchaseSpecDiagnosticSource(db *sql.DB, goodsID string) string {
product, err := repository.GetPddProductByGoodsID(db, goodsID)
if err != nil || product == nil || strings.TrimSpace(product.SkusJSON) == "" {
return ""
}
var metadata struct {
SpecSource string `json:"spec_source"`
}
if json.Unmarshal([]byte(product.SkusJSON), &metadata) != nil {
return ""
}
return strings.TrimSpace(metadata.SpecSource)
}
var runtimeWeightPattern = regexp.MustCompile(`(?i)^\s*(\d+(?:\.\d+)?)\s*(?:[-~~—–至到]\s*(\d+(?:\.\d+)?))?\s*(公斤|kg|kgs|千克|斤)\s*$`)
type runtimeWeightRange struct{ fromJin, toJin float64 }
func matchPurchaseSpecByRule(target string, candidates []PurchaseSpecCandidate) (purchaseSpecDecision, bool) {
matches := make([]PurchaseSpecCandidate, 0, 1)
for _, candidate := range candidates {
if runtimeSpecsEquivalent(target, candidate.RawText) {
matches = append(matches, candidate)
}
}
if len(matches) == 1 {
return purchaseSpecDecision{Outcome: model.PurchaseSpecResolutionMatched,
Source: model.PurchaseSpecResolutionRule, Candidate: &matches[0],
ConfidenceBPS: 10000, ConfidenceSet: true,
Reason: "目标尺码与唯一真机候选按确定性规则等价"}, true
}
if len(matches) > 1 {
return purchaseSpecDecision{Outcome: model.PurchaseSpecResolutionUncertain,
Source: model.PurchaseSpecResolutionRule,
Reason: "存在多个与目标尺码等价的真机候选,不能安全自动选择"}, true
}
// 两个重量区间同时覆盖目标时,模型也没有可靠信号判断页面上的业务边界,
// 直接返回不确定,避免让语言模型替代真实尺码规则猜一个。
if targetWeight, ok := parseRuntimeWeightRange(target); ok {
overlaps := 0
for _, candidate := range candidates {
if candidateWeight, candidateOK := parseRuntimeWeightRange(candidate.RawText); candidateOK &&
weightRangesOverlap(targetWeight, candidateWeight) {
overlaps++
}
}
if overlaps > 1 {
return purchaseSpecDecision{Outcome: model.PurchaseSpecResolutionUncertain,
Source: model.PurchaseSpecResolutionRule,
Reason: "目标重量区间同时落入多个真机候选,不能安全自动选择"}, true
}
}
return purchaseSpecDecision{}, false
}
func runtimeSpecsEquivalent(left, right string) bool {
a, okA := parseRuntimeWeightRange(left)
b, okB := parseRuntimeWeightRange(right)
if okA || okB {
return okA && okB && almostEqual(a.fromJin, b.fromJin) && almostEqual(a.toJin, b.toJin)
}
return normalizeRuntimeSpec(left) == normalizeRuntimeSpec(right)
}
func normalizeRuntimeSpec(raw string) string {
raw = simplifyExplicit(raw)
for _, pair := range []struct{ from, to string }{
{"體", "体"}, {"號", "号"}, {"圍", "围"}, {"寬", "宽"}, {"適", "适"},
{"齡", "龄"}, {"歲", "岁"}, {"單", "单"}, {"雙", "双"}, {"釐", "厘"},
{"臺", "台"}, {"兒", "儿"}, {"婦", "妇"}, {"標", "标"}, {"準", "准"},
{"顏", "颜"}, {"議", "议"}, {"鬆", "松"}, {"褲", "裤"}, {"絨", "绒"},
{"寶", "宝"}, {"嬰", "婴"},
} {
raw = strings.ReplaceAll(raw, pair.from, pair.to)
}
var result strings.Builder
for _, r := range strings.ToLower(raw) {
// 加减号和范围符号可能是尺码语义(例如 M+、50-60kg),不能为了
// “忽略符号”把两个不同规格折叠成同一个。斜杠、括号、冒号等展示
// 分隔符才可以安全忽略。
if strings.ContainsRune("+-~~—–至到", r) {
result.WriteRune(r)
continue
}
if unicode.IsSpace(r) || unicode.IsPunct(r) || unicode.IsSymbol(r) {
continue
}
result.WriteRune(r)
}
return result.String()
}
func parseRuntimeWeightRange(raw string) (runtimeWeightRange, bool) {
raw = strings.TrimFunc(raw, func(r rune) bool {
return (unicode.IsPunct(r) || unicode.IsSymbol(r)) && !strings.ContainsRune("+-~~—–", r)
})
match := runtimeWeightPattern.FindStringSubmatch(simplifyExplicit(raw))
if len(match) == 0 {
return runtimeWeightRange{}, false
}
from, err := strconv.ParseFloat(match[1], 64)
if err != nil || from <= 0 {
return runtimeWeightRange{}, false
}
to := from
if match[2] != "" {
to, err = strconv.ParseFloat(match[2], 64)
if err != nil || to <= 0 || from > to {
return runtimeWeightRange{}, false
}
}
if unit := strings.ToLower(match[3]); unit == "公斤" || unit == "kg" || unit == "kgs" || unit == "千克" {
from *= 2
to *= 2
}
return runtimeWeightRange{fromJin: from, toJin: to}, true
}
func weightRangesOverlap(left, right runtimeWeightRange) bool {
return left.fromJin <= right.toJin && right.fromJin <= left.toJin
}
func almostEqual(left, right float64) bool {
delta := left - right
if delta < 0 {
delta = -delta
}
return delta < 0.000001
}
func matchPurchaseSpecWithAI(ctx context.Context, db *sql.DB, secrets AISecretStore,
policy AIEndpointPolicy, req PurchaseSpecResolutionRequest, loadSnapshot aiSnapshotLoader) purchaseSpecDecision {
snapshot, err := loadSnapshot(ctx, db, secrets, policy)
if err != nil {
return purchaseSpecDecision{Outcome: model.PurchaseSpecResolutionFailed,
Reason: truncateRunes("AI 运行配置不可用:"+safeAIError(err), 500)}
}
base := purchaseSpecDecision{Source: model.PurchaseSpecResolutionAI,
ProviderID: snapshot.Provider.ProviderID, SourceModel: snapshot.Provider.Model,
Fingerprint: snapshot.ConfigFingerprint, PromptVersion: AISpecMatchPromptVersion}
request := AIModelMatchRequest{
SourceSpec: "已选择颜色:" + truncateRunes(req.SelectedColor, 191) + ";任务目标尺码:" + truncateRunes(req.TargetSize, 191),
}
for _, candidate := range req.Candidates {
request.Candidates = append(request.Candidates, AIModelCandidate{
ID: candidate.CandidateID, Label: candidate.RawText, Options: boundedAIOptions(candidate.Options),
})
}
selection := selectAIWhitelistedCandidate(ctx, snapshot, request)
response := selection.Response
base.ConfidenceBPS, base.ConfidenceSet, base.Reason = response.ConfidenceBPS, selection.ConfidenceSet, selection.Reason
switch selection.Status {
case aiSelectionFailed:
base.Outcome = model.PurchaseSpecResolutionFailed
return base
case aiSelectionUncertain, aiSelectionDimensionsUnresolved, aiSelectionBelowThreshold:
base.Outcome = model.PurchaseSpecResolutionUncertain
return base
case aiSelectionConflict, aiSelectionCandidateRejected:
base.Outcome = model.PurchaseSpecResolutionRejected
return base
}
var selected *PurchaseSpecCandidate
for index := range req.Candidates {
if req.Candidates[index].CandidateID == response.CandidateID {
selected = &req.Candidates[index]
break
}
}
if selected == nil {
base.Outcome, base.Reason = model.PurchaseSpecResolutionRejected, "候选绑定已变化,未返回匹配"
return base
}
base.Outcome, base.Candidate = model.PurchaseSpecResolutionMatched, selected
return base
}
func completePurchaseSpecResolution(db *sql.DB, idempotencyKey, requestHash string,
resolution *model.PurchaseSpecResolution, candidates []PurchaseSpecCandidate,
result purchaseSpecDecision) (string, error) {
tx, err := db.Begin()
if err != nil {
return "", fmt.Errorf("%w: 开始保存解析结论事务失败", ErrSpecResolutionUnavailable)
}
defer tx.Rollback()
if body, done, lookupErr := repository.LookupIdempotent(tx, idempotencyKey, requestHash); lookupErr != nil {
if errors.Is(lookupErr, repository.ErrIdempotencyConflict) {
return "", ErrIdempotencyConflict
}
return "", fmt.Errorf("%w: 查询幂等结果失败", ErrSpecResolutionUnavailable)
} else if done {
return body, nil
}
decidedAt := model.NowISO()
decision := model.PurchaseSpecResolutionDecision{
ResolutionID: resolution.ResolutionID, Outcome: result.Outcome, DecisionSource: result.Source,
ConfidenceBPS: result.ConfidenceBPS, ConfidenceSet: result.ConfidenceSet,
Reason: truncateRunes(result.Reason, 500), ProviderID: result.ProviderID,
SourceModel: result.SourceModel, ConfigFingerprint: result.Fingerprint,
RulesVersion: PurchaseSpecRulesVersion, PromptVersion: result.PromptVersion, DecidedAt: decidedAt,
}
if result.Candidate != nil {
decision.ChosenCandidateID = result.Candidate.CandidateID
resolvedJSON, _ := json.Marshal(result.Candidate.Options)
decision.ResolvedOptionsJSON = string(resolvedJSON)
}
if err := repository.CompletePurchaseSpecResolution(tx, decision); err != nil {
if errors.Is(err, repository.ErrPurchaseSpecResolutionCompleted) {
if rollbackErr := tx.Rollback(); rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
return "", fmt.Errorf("回滚并发规格解析失败: %w", rollbackErr)
}
existing, readErr := repository.GetPurchaseSpecResolutionForReplay(
db, resolution.TaskID, resolution.AttemptID, resolution.CandidateSnapshotHash, requestHash)
if readErr != nil || existing == nil || existing.Outcome == model.PurchaseSpecResolutionPending {
return "", fmt.Errorf("%w: 读取并发解析结论失败", ErrSpecResolutionUnavailable)
}
return saveReconstructedPurchaseSpecResponse(db, idempotencyKey, requestHash, existing, candidates)
}
return "", fmt.Errorf("%w: 保存解析结论失败", ErrSpecResolutionUnavailable)
}
response := purchaseSpecResponse(resolution.ResolutionID, resolution.CandidateSnapshotHash,
decidedAt, result.Outcome, result.Source, result.Candidate, result.ConfidenceBPS,
result.ConfidenceSet, decision.Reason)
body, err := json.Marshal(response)
if err != nil {
return "", fmt.Errorf("序列化规格解析响应失败: %w", err)
}
if err := repository.SaveIdempotent(tx, idempotencyKey, requestHash, string(body)); err != nil {
if errors.Is(err, repository.ErrIdempotencyAlreadySaved) {
if rollbackErr := tx.Rollback(); rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {
return "", fmt.Errorf("回滚重复规格解析失败: %w", rollbackErr)
}
replayed, done, lookupErr := repository.LookupIdempotent(db, idempotencyKey, requestHash)
if lookupErr != nil || !done {
return "", fmt.Errorf("%w: 读取并发幂等响应失败", ErrSpecResolutionUnavailable)
}
return replayed, nil
}
return "", fmt.Errorf("%w: 保存规格解析幂等响应失败", ErrSpecResolutionUnavailable)
}
if err := repository.TouchClient(tx, resolution.ClientID); err != nil {
return "", fmt.Errorf("%w: 刷新客户端活动时间失败", ErrSpecResolutionUnavailable)
}
if err := tx.Commit(); err != nil {
return "", fmt.Errorf("%w: 提交规格解析结论失败", ErrSpecResolutionUnavailable)
}
return string(body), nil
}
func saveReconstructedPurchaseSpecResponse(db *sql.DB, idempotencyKey, requestHash string,
resolution *model.PurchaseSpecResolution, candidates []PurchaseSpecCandidate) (string, error) {
var candidate *PurchaseSpecCandidate
for index := range candidates {
if candidates[index].CandidateID == resolution.ChosenCandidateID {
candidate = &candidates[index]
break
}
}
if resolution.Outcome == model.PurchaseSpecResolutionMatched && candidate == nil {
return "", fmt.Errorf("%w: 已保存结论的候选编号不在请求快照中", ErrSpecResolutionUnavailable)
}
response := purchaseSpecResponse(resolution.ResolutionID, resolution.CandidateSnapshotHash,
resolution.DecidedAt, resolution.Outcome, resolution.DecisionSource, candidate,
resolution.ConfidenceBPS, resolution.ConfidenceSet, resolution.Reason)
body, err := json.Marshal(response)
if err != nil {
return "", err
}
tx, err := db.Begin()
if err != nil {
return "", fmt.Errorf("%w: 开始恢复幂等响应事务失败", ErrSpecResolutionUnavailable)
}
defer tx.Rollback()
if replayed, done, lookupErr := repository.LookupIdempotent(tx, idempotencyKey, requestHash); lookupErr != nil {
if errors.Is(lookupErr, repository.ErrIdempotencyConflict) {
return "", ErrIdempotencyConflict
}
return "", lookupErr
} else if done {
return replayed, nil
}
if err := repository.SaveIdempotent(tx, idempotencyKey, requestHash, string(body)); err != nil {
return "", err
}
if err := tx.Commit(); err != nil {
return "", err
}
return string(body), nil
}
func purchaseSpecResponse(resolutionID, snapshotHash, resolvedAt string,
outcome model.PurchaseSpecResolutionOutcome, source model.PurchaseSpecResolutionSource,
candidate *PurchaseSpecCandidate, confidence int, confidenceSet bool, reason string) PurchaseSpecResolutionResponse {
response := PurchaseSpecResolutionResponse{SchemaVersion: purchaseSpecSchemaVersion,
ResolutionID: resolutionID, Outcome: outcome, CandidateSnapshotHash: snapshotHash,
Reason: reason, ResolvedAt: resolvedAt}
if source != "" {
value := source
response.Source = &value
}
if confidenceSet {
value := confidence
response.ConfidenceBPS = &value
}
if candidate != nil {
response.Match = &PurchaseSpecResolutionMatch{CandidateID: candidate.CandidateID,
RawText: candidate.RawText, Options: candidate.Options}
}
return response
}
@@ -0,0 +1,440 @@
package service
import (
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"os"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"cmautobuy/admin/model"
"cmautobuy/admin/repository"
)
func TestMatchPurchaseSpecByRule_只接受唯一等价候选(t *testing.T) {
tests := []struct {
name string
target string
values []string
terminal bool
outcome model.PurchaseSpecResolutionOutcome
candidate string
}{
{name: "逐字相同", target: "M码", values: []string{"S码", "M码"}, terminal: true,
outcome: model.PurchaseSpecResolutionMatched, candidate: "c2"},
{name: "繁简大小写和展示符号", target: "藍 色 / XL 碼", values: []string{"蓝色/xl码"}, terminal: true,
outcome: model.PurchaseSpecResolutionMatched, candidate: "c1"},
{name: "公斤和斤单值", target: "60公斤", values: []string{"100斤", "120斤"}, terminal: true,
outcome: model.PurchaseSpecResolutionMatched, candidate: "c2"},
{name: "公斤和斤正区间", target: "50-60 kg", values: []string{"80~100斤", "100至120斤"}, terminal: true,
outcome: model.PurchaseSpecResolutionMatched, candidate: "c2"},
{name: "多个重叠区间直接不确定", target: "50-60公斤", values: []string{"90-110斤", "110-130斤"}, terminal: true,
outcome: model.PurchaseSpecResolutionUncertain},
{name: "倒序区间不自动修正", target: "60-50公斤", values: []string{"100-120斤"}, terminal: false},
{name: "加减尺码不能被符号归一化合并", target: "M+", values: []string{"M-"}, terminal: false},
{name: "多个等价候选不选一个", target: "60公斤", values: []string{"120斤", "60kg"}, terminal: true,
outcome: model.PurchaseSpecResolutionUncertain},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
candidates := runtimeCandidates("黑色", test.values...)
decision, terminal := matchPurchaseSpecByRule(test.target, candidates)
if terminal != test.terminal || decision.Outcome != test.outcome {
t.Fatalf("terminal=%v outcome=%s", terminal, decision.Outcome)
}
if test.candidate != "" && (decision.Candidate == nil || decision.Candidate.CandidateID != test.candidate) {
t.Fatalf("candidate=%+v", decision.Candidate)
}
if test.outcome != model.PurchaseSpecResolutionMatched && decision.Candidate != nil {
t.Fatalf("非 matched 不应返回候选: %+v", decision.Candidate)
}
})
}
}
func TestParsePurchaseSpecResolutionRequest_校验候选快照和确定性幂等键(t *testing.T) {
req, body, key := validRuntimeSpecRequest(t, "cg255", "M码", "S码", "M码")
parsed, err := parsePurchaseSpecResolutionRequest("cg255", key, body)
if err != nil || parsed.CandidateSnapshotHash != req.CandidateSnapshotHash {
t.Fatalf("合法请求解析失败: parsed=%+v err=%v", parsed, err)
}
badHash := append([]byte(nil), body...)
var changed map[string]any
if err := json.Unmarshal(badHash, &changed); err != nil {
t.Fatal(err)
}
changed["candidate_snapshot_hash"] = "0000000000000000000000000000000000000000000000000000000000000000"
badHash, _ = json.Marshal(changed)
if _, err := parsePurchaseSpecResolutionRequest("cg255", key, badHash); !errors.Is(err, ErrSpecResolutionHashMismatch) {
t.Fatalf("错误快照哈希 err=%v", err)
}
req.Candidates[0].CandidateID = "c2"
req.CandidateSnapshotHash = purchaseSpecCandidateSnapshotHash(req)
invalidBody, _ := json.Marshal(req)
invalidKey := purchaseSpecIdempotencyKey("cg255", req)
if _, err := parsePurchaseSpecResolutionRequest("cg255", invalidKey, invalidBody); !errors.Is(err, ErrInvalidSpecResolutionRequest) {
t.Fatalf("跳号候选 err=%v", err)
}
tooLarge := make([]byte, MaxPurchaseSpecResolutionBodyBytes+1)
if _, err := parsePurchaseSpecResolutionRequest("cg255", key, tooLarge); !errors.Is(err, ErrInvalidSpecResolutionBody) {
t.Fatalf("超大请求 err=%v", err)
}
}
type stubAIModelClient struct {
response AIModelMatchResponse
err error
delay time.Duration
calls atomic.Int32
last AIModelMatchRequest
mu sync.Mutex
}
func (client *stubAIModelClient) Match(ctx context.Context, _ model.AIProviderConfig, _ string,
request AIModelMatchRequest) (AIModelMatchResponse, error) {
client.calls.Add(1)
client.mu.Lock()
client.last = request
client.mu.Unlock()
if client.delay > 0 {
select {
case <-time.After(client.delay):
case <-ctx.Done():
return AIModelMatchResponse{}, ctx.Err()
}
}
return client.response, client.err
}
func TestMatchPurchaseSpecWithAI_只接受请求候选并执行安全门禁(t *testing.T) {
tests := []struct {
name string
response AIModelMatchResponse
modelErr error
threshold int
outcome model.PurchaseSpecResolutionOutcome
candidate string
}{
{name: "白名单内高置信命中", threshold: 8000, outcome: model.PurchaseSpecResolutionMatched, candidate: "c2",
response: AIModelMatchResponse{Conclusion: "match", CandidateID: "c2", ConfidenceBPS: 9200, Reason: "唯一对应"}},
{name: "越权候选拒绝", threshold: 8000, outcome: model.PurchaseSpecResolutionRejected,
response: AIModelMatchResponse{Conclusion: "match", CandidateID: "invented", ConfidenceBPS: 9200, Reason: "错误编号"}},
{name: "低于阈值不自动选择", threshold: 9500, outcome: model.PurchaseSpecResolutionUncertain,
response: AIModelMatchResponse{Conclusion: "match", CandidateID: "c2", ConfidenceBPS: 9200, Reason: "置信不足"}},
{name: "冲突维度不自动选择", threshold: 8000, outcome: model.PurchaseSpecResolutionUncertain,
response: AIModelMatchResponse{Conclusion: "match", CandidateID: "c2", ConfidenceBPS: 9200, Reason: "仍缺信息", MissingDimensions: []string{"尺码"}}},
{name: "模型格式错误记失败", threshold: 8000, outcome: model.PurchaseSpecResolutionFailed,
response: AIModelMatchResponse{Conclusion: "other", ConfidenceBPS: 9200, Reason: "非法"}},
{name: "模型超时和密钥不进入原因", threshold: 8000, outcome: model.PurchaseSpecResolutionFailed,
modelErr: errors.New("upstream timeout for secret")},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
client := &stubAIModelClient{response: test.response, err: test.modelErr}
loader := func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
return AIMatchSnapshot{Provider: model.AIProviderConfig{ProviderID: "provider-1", Model: "model-1",
ConfidenceThresholdBPS: test.threshold}, ConfigFingerprint: "fingerprint", Secret: "secret", Client: client}, nil
}
req, _, _ := validRuntimeSpecRequest(t, "cg255", "L码", "M码", "XL码")
decision := matchPurchaseSpecWithAI(context.Background(), nil, nil, AIEndpointPolicy{}, req, loader)
if decision.Outcome != test.outcome {
t.Fatalf("outcome=%s reason=%s", decision.Outcome, decision.Reason)
}
if test.modelErr != nil && strings.Contains(decision.Reason, "secret") {
t.Fatalf("模型错误泄露密钥: %s", decision.Reason)
}
if test.candidate != "" && (decision.Candidate == nil || decision.Candidate.CandidateID != test.candidate) {
t.Fatalf("candidate=%+v", decision.Candidate)
}
if test.outcome != model.PurchaseSpecResolutionMatched && decision.Candidate != nil {
t.Fatalf("非 matched 不得带候选: %+v", decision.Candidate)
}
if client.calls.Load() != 1 {
t.Fatalf("model calls=%d", client.calls.Load())
}
for index, candidate := range client.last.Candidates {
if candidate.ID != fmt.Sprintf("c%d", index+1) {
t.Fatalf("AI 候选编号被改写: %+v", client.last.Candidates)
}
}
})
}
}
func TestResolvePurchaseSpec_规则命中幂等且不改任务主数据(t *testing.T) {
db := newRuntimeSpecTestDB(t)
prepareRuntimeSpecTask(t, db, "cg255-rule", model.TaskPurchase, true, `{"color":"黑色","size":"60公斤"}`)
now := model.NowISO()
originalSKUs := `{"spec_source":"client","skus":[{"options":{"color":"黑色","size":"60公斤"},"available":true}]}`
if _, err := db.Exec(`INSERT INTO pdd_products
(goods_id,url,title,skus_json,collect_status,created_at,updated_at)
VALUES(?,?,?,?,'collected',?,?)`, "PDD-255", "https://mobile.yangkeduo.com/goods.html?goods_id=PDD-255",
"测试商品", originalSKUs, now, now); err != nil {
t.Fatal(err)
}
_, body, key := validRuntimeSpecRequest(t, "cg255-rule", "60公斤", "100斤", "120斤")
var loaderCalls atomic.Int32
loader := func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
loaderCalls.Add(1)
return AIMatchSnapshot{}, errors.New("规则命中不应加载 AI")
}
first, err := resolvePurchaseSpecWithSnapshotLoader(context.Background(), db, nil, AIEndpointPolicy{},
"cg255-rule", "CLIENT-255", key, body, loader)
if err != nil {
t.Fatal(err)
}
second, err := resolvePurchaseSpecWithSnapshotLoader(context.Background(), db, nil, AIEndpointPolicy{},
"cg255-rule", "CLIENT-255", key, body, loader)
if err != nil || second != first {
t.Fatalf("幂等重放不同: first=%s second=%s err=%v", first, second, err)
}
if loaderCalls.Load() != 0 {
t.Fatalf("规则命中却调用 AI %d 次", loaderCalls.Load())
}
var response PurchaseSpecResolutionResponse
if err := json.Unmarshal([]byte(first), &response); err != nil {
t.Fatal(err)
}
if response.Outcome != model.PurchaseSpecResolutionMatched || response.Source == nil ||
*response.Source != model.PurchaseSpecResolutionRule || response.Match == nil || response.Match.CandidateID != "c2" {
t.Fatalf("响应错误: %+v", response)
}
var status, options string
var version, maxPrice int64
var resultData sql.NullString
if err := db.QueryRow(`SELECT status,version,pdd_options,max_price_cent,result_data FROM tasks WHERE task_id=?`, "cg255-rule").
Scan(&status, &version, &options, &maxPrice, &resultData); err != nil {
t.Fatal(err)
}
if status != string(model.TaskClaimed) || version != 1 || options != `{"color":"黑色","size":"60公斤"}` ||
maxPrice != 2000 || resultData.Valid {
t.Fatalf("任务被规格解析改写: status=%s version=%d options=%s max=%d result=%q",
status, version, options, maxPrice, resultData.String)
}
var actualSKUs string
if err := db.QueryRow(`SELECT skus_json FROM pdd_products WHERE goods_id='PDD-255'`).Scan(&actualSKUs); err != nil || actualSKUs != originalSKUs {
t.Fatalf("PDD 主规格被改写: skus=%s err=%v", actualSKUs, err)
}
var count int
if err := db.QueryRow(`SELECT COUNT(*) FROM purchase_spec_resolutions WHERE task_id=?`, "cg255-rule").Scan(&count); err != nil || count != 1 {
t.Fatalf("审计记录 count=%d err=%v", count, err)
}
}
func TestResolvePurchaseSpec_业务权限错误不写解析记录(t *testing.T) {
tests := []struct {
name string
taskType model.TaskType
claimed bool
mutate func(*PurchaseSpecResolutionRequest)
expected error
}{
{name: "非采购任务", taskType: model.TaskCollect, claimed: true, expected: ErrTaskNotPurchase},
{name: "从未领取", taskType: model.TaskPurchase, claimed: false, expected: ErrNeverClaimed},
{name: "任务版本冲突", taskType: model.TaskPurchase, claimed: true,
mutate: func(req *PurchaseSpecResolutionRequest) { req.TaskVersion = 2 }, expected: ErrTaskVersionConflict},
{name: "PDD 商品不一致", taskType: model.TaskPurchase, claimed: true,
mutate: func(req *PurchaseSpecResolutionRequest) { req.PddGoodsID = "OTHER" }, expected: ErrPDDGoodsMismatch},
{name: "任务原始规格被改写", taskType: model.TaskPurchase, claimed: true,
mutate: func(req *PurchaseSpecResolutionRequest) { req.OriginalOptions["size"] = "XL码" }, expected: ErrInvalidSpecResolutionRequest},
}
for index, test := range tests {
t.Run(test.name, func(t *testing.T) {
db := newRuntimeSpecTestDB(t)
taskID := fmt.Sprintf("cg255-reject-%d", index)
prepareRuntimeSpecTask(t, db, taskID, test.taskType, test.claimed, `{"color":"黑色","size":"M码"}`)
req, _, _ := validRuntimeSpecRequest(t, taskID, "M码", "M码", "L码")
if test.mutate != nil {
test.mutate(&req)
}
req.CandidateSnapshotHash = purchaseSpecCandidateSnapshotHash(req)
body, _ := json.Marshal(req)
key := purchaseSpecIdempotencyKey(taskID, req)
_, err := resolvePurchaseSpecWithSnapshotLoader(context.Background(), db, nil, AIEndpointPolicy{},
taskID, "CLIENT-255", key, body, func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
t.Fatal("权限错误前不应调用 AI")
return AIMatchSnapshot{}, nil
})
if !errors.Is(err, test.expected) {
t.Fatalf("err=%v expected=%v", err, test.expected)
}
var count int
if queryErr := db.QueryRow(`SELECT COUNT(*) FROM purchase_spec_resolutions`).Scan(&count); queryErr != nil || count != 0 {
t.Fatalf("拒绝请求产生审计记录 count=%d err=%v", count, queryErr)
}
})
}
}
func TestResolvePurchaseSpec_AI并发重放只产生一个模型调用和一个结论(t *testing.T) {
db := newRuntimeSpecTestDB(t)
prepareRuntimeSpecTask(t, db, "cg255-ai", model.TaskPurchase, true, `{"color":"黑色","size":"L码"}`)
_, body, key := validRuntimeSpecRequest(t, "cg255-ai", "L码", "M码", "XL码")
client := &stubAIModelClient{delay: 50 * time.Millisecond,
response: AIModelMatchResponse{Conclusion: "match", CandidateID: "c2", ConfidenceBPS: 9300, Reason: "模型唯一匹配"}}
loader := func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
return AIMatchSnapshot{Provider: model.AIProviderConfig{ProviderID: "provider-1", Model: "model-1",
ConfidenceThresholdBPS: 9000}, ConfigFingerprint: "fingerprint", Secret: "secret", Client: client}, nil
}
const workers = 6
responses := make(chan string, workers)
errorsFound := make(chan error, workers)
var wait sync.WaitGroup
for range workers {
wait.Add(1)
go func() {
defer wait.Done()
response, err := resolvePurchaseSpecWithSnapshotLoader(context.Background(), db, nil, AIEndpointPolicy{},
"cg255-ai", "CLIENT-255", key, body, loader)
responses <- response
errorsFound <- err
}()
}
wait.Wait()
close(responses)
close(errorsFound)
for err := range errorsFound {
if err != nil {
t.Fatal(err)
}
}
first := ""
for response := range responses {
if first == "" {
first = response
} else if response != first {
t.Fatalf("并发响应不一致: first=%s other=%s", first, response)
}
}
if client.calls.Load() != 1 {
t.Fatalf("模型调用次数=%d", client.calls.Load())
}
var records, finalRecords, idempotency int
if err := db.QueryRow(`SELECT COUNT(*),SUM(outcome<>'pending') FROM purchase_spec_resolutions WHERE task_id=?`, "cg255-ai").Scan(&records, &finalRecords); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT COUNT(*) FROM idempotency_keys WHERE `+"`key`"+`=?`, key).Scan(&idempotency); err != nil {
t.Fatal(err)
}
if records != 1 || finalRecords != 1 || idempotency != 1 {
t.Fatalf("records=%d final=%d idempotency=%d", records, finalRecords, idempotency)
}
}
func TestResolvePurchaseSpec_ShopeeBackfill只作诊断不阻断规则(t *testing.T) {
db := newRuntimeSpecTestDB(t)
prepareRuntimeSpecTask(t, db, "cg255-backfill", model.TaskPurchase, true, `{"color":"黑色","size":"60公斤"}`)
now := model.NowISO()
_, err := db.Exec(`INSERT INTO pdd_products
(goods_id,url,title,skus_json,collect_status,created_at,updated_at)
VALUES(?,?,?,?,'collected',?,?)`, "PDD-255", "https://mobile.yangkeduo.com/goods.html?goods_id=PDD-255",
"测试商品", `{"spec_source":"shopee_backfill","skus":[]}`, now, now)
if err != nil {
t.Fatal(err)
}
_, body, key := validRuntimeSpecRequest(t, "cg255-backfill", "60公斤", "120斤")
response, err := resolvePurchaseSpecWithSnapshotLoader(context.Background(), db, nil, AIEndpointPolicy{},
"cg255-backfill", "CLIENT-255", key, body, func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
t.Fatal("规则唯一命中不应调用 AI")
return AIMatchSnapshot{}, nil
})
if err != nil {
t.Fatal(err)
}
var parsed PurchaseSpecResolutionResponse
if err := json.Unmarshal([]byte(response), &parsed); err != nil || parsed.Outcome != model.PurchaseSpecResolutionMatched {
t.Fatalf("response=%s err=%v", response, err)
}
if !strings.Contains(parsed.Reason, "诊断来源:shopee_backfill") {
t.Fatalf("未保存回填规格诊断来源: %+v", parsed)
}
}
func runtimeCandidates(color string, values ...string) []PurchaseSpecCandidate {
result := make([]PurchaseSpecCandidate, 0, len(values))
for index, value := range values {
result = append(result, PurchaseSpecCandidate{CandidateID: fmt.Sprintf("c%d", index+1), RawText: value,
Options: map[string]string{"color": color, "size": value}})
}
return result
}
func validRuntimeSpecRequest(t *testing.T, taskID, target string, candidates ...string) (PurchaseSpecResolutionRequest, []byte, string) {
t.Helper()
req := PurchaseSpecResolutionRequest{SchemaVersion: 1, TaskVersion: 1, AttemptID: "attempt-255",
PddGoodsID: "PDD-255", OriginalOptions: map[string]string{"color": "黑色", "size": target},
SelectedColor: "黑色", TargetSize: target, Candidates: runtimeCandidates("黑色", candidates...),
ObservedAt: "2026-08-17T08:00:00Z"}
req.CandidateSnapshotHash = purchaseSpecCandidateSnapshotHash(req)
body, err := json.Marshal(req)
if err != nil {
t.Fatal(err)
}
return req, body, purchaseSpecIdempotencyKey(taskID, req)
}
func prepareRuntimeSpecTask(t *testing.T, db *sql.DB, taskID string, taskType model.TaskType, claimed bool, options string) {
t.Helper()
now := model.NowISO()
_, err := db.Exec(`INSERT INTO clients
(client_id,name,device_address,platform,pdd_package,capabilities,last_seen_at,created_at,updated_at)
VALUES('CLIENT-255','测试客户端','','android','','{}',?,?,?)`, now, now, now)
if err != nil {
t.Fatal(err)
}
_, err = db.Exec(`INSERT INTO tasks
(task_id,task_type,status,version,assigned_client,goods_id,pdd_goods_url,pdd_goods_id,pdd_options,
quantity,max_price_cent,created_at,updated_at)
VALUES(?,?,?,1,'CLIENT-255','SHOPEE-255',?,?,?,1,2000,?,?)`, taskID, taskType, model.TaskClaimed,
"https://mobile.yangkeduo.com/goods.html?goods_id=PDD-255", "PDD-255", options, now, now)
if err != nil {
t.Fatal(err)
}
if claimed {
if _, err := db.Exec(`INSERT INTO task_claims(task_id,client_id,claimed_at) VALUES(?,?,?)`,
taskID, "CLIENT-255", now); err != nil {
t.Fatal(err)
}
}
}
// newRuntimeSpecTestDB 默认使用历史 SQLite 测试库跑服务编排;显式打开 MySQL 测试时
// 自动切到隔离的 MySQL 8 数据库。生产代码仍只运行在 MySQL。
func newRuntimeSpecTestDB(t *testing.T) *sql.DB {
t.Helper()
if os.Getenv("CMAUTOBUY_MYSQL_TEST") == "1" {
return newTestDB(t)
}
db, err := repository.Open(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { db.Close() })
if err := repository.Migrate(db); err != nil {
t.Fatal(err)
}
_, err = db.Exec(`CREATE TABLE purchase_spec_resolutions (
resolution_id TEXT PRIMARY KEY,task_id TEXT NOT NULL,attempt_id TEXT NOT NULL,client_id TEXT NOT NULL,
task_version INTEGER NOT NULL,pdd_goods_id TEXT NOT NULL,original_options_json TEXT NOT NULL,
selected_color TEXT NOT NULL,target_size TEXT NOT NULL,candidates_json TEXT NOT NULL,
candidate_snapshot_hash TEXT NOT NULL,request_hash TEXT NOT NULL,observed_at TEXT NOT NULL,
outcome TEXT NOT NULL DEFAULT 'pending',decision_source TEXT,chosen_candidate_id TEXT,
resolved_options_json TEXT,confidence_bps INTEGER,reason TEXT,provider_id TEXT,source_model TEXT,
config_fingerprint TEXT,rules_version TEXT,prompt_version TEXT,created_at TEXT NOT NULL,decided_at TEXT,
UNIQUE(task_id,attempt_id,candidate_snapshot_hash)
)`)
if err != nil {
t.Fatal(err)
}
return db
}
+26
View File
@@ -415,6 +415,32 @@ Admin 本地时区,付款状态只是 Client 核单上报时的快照,Admin
- “AI规格匹配”是当前有效映射的来源筛选,不替代“可创建采购任务”等业务阶段。人工修改后
当前来源立即变为人工,但历史 AI 决策和批次记录继续保留。
### 4.7 采购运行时真机规格解析(#255)
采购任务已经在 Client 上选定颜色,但任务目标尺码和当前 PDD 页面文字无法精确对应时,
Client 可以把这一刻页面显示的可购买尺码提交给 Admin 做一次解析。该能力只用于解决
“任务规格与真机候选文字不一致”,不会改变正常精确匹配流程。
- Admin 必须先核对采购任务、任务版本、PDD 商品、任务原始规格和该 Client 的领取历史;
任一身份不一致都拒绝,并且不产生解析决策。
- 候选编号、原文、颜色和尺码由 Client 按页面顺序固定。Admin 重新计算候选快照哈希,
AI 只能选择本次请求内的候选短编号,不能生成或改写 PDD 规格。
- 先执行确定性规则:忽略大小写、空白和分隔符,处理已明确的繁简差异,并支持唯一的
`kg / 公斤 / 千克 / 斤` 单值或正向区间等价。只有唯一候选等价时才自动返回;多个等价、
倒序或多个候选同时重叠时直接返回 `uncertain`,不交给 AI 猜一个。
- 规则没有唯一结论时才使用当前已启用且通过连接测试的 AI 配置;模型结论仍须通过候选
白名单、冲突/缺失维度和管理员置信度阈值门禁。模型超时、格式错误或越权候选都形成
可审计结论,不把模型输出直接写进任务。
- `pdd_products.skus_json.spec_source=shopee_backfill` 只用于诊断规格来源,既不自动触发 AI,
也不拒绝本次解析;是否需要解析只由真机候选和任务目标能否唯一对应决定。
- 相同任务、执行尝试和候选快照幂等重放时返回第一次的完整响应;并发相同请求只完成一条
最终决策。候选观察与最终决策分两个短事务,调用模型期间不持有数据库事务。
- 解析记录是只追加的运行审计。它不得修改 `tasks` 状态、版本、目标规格、价格、
`task_runs.irreversible_action_at` 或 `pdd_products.skus_json`,也不得保存控件树、截图、订单、
收货信息、Cookie、Token 或 API Key。
- Admin 返回候选后,Client 仍须重新读取页面并继续执行商品、规格、数量、总价、地址、
不可逆标记和单次提交门禁;运行时解析不能绕过任何真实下单安全检查。
## 5. 创建采购任务的校验
`[必须]` 下面任何一条不满足就不允许创建,并明确告诉操作员缺什么:
+17
View File
@@ -248,6 +248,8 @@ handler 返回批次 JSON;管理员在统一导入记录页查看摘要
- Admin 是任务的创建者和分配者,Client 是执行者。
- Admin **不知道** Client 执行到哪一步(没有心跳,是有意的)。
- Client 提交结果时,Admin **无条件接受**,哪怕任务已取消或已重派。
- Client 的运行时规格解析是 `POST /api/v1/client/tasks/{task_id}/spec-resolution` 一次性命令,
不是任务状态查询或轮询;权威字段、哈希和错误码仍以 Client 侧契约 §7.1 为准。
- Admin 用户登录与 Client 身份是两套边界。Web Session 不传给 Client,
Web 登录中间件也不覆盖 `/api/v1/client/*`。
- 详见 [04 Client 接口实现](04-client-api.md)。
@@ -269,6 +271,21 @@ URL 内凭据、环回、链路本地、云元数据和未显式允许的私网
唯一确定的规则结果不调用模型;有效人工映射始终优先。保存前重新计算上下文版本并检查
当前可购买候选,防止 PDD 重采集或人工并发修改后写入过期结果。
采购运行时规格解析复用同一个 `AISecretStore`、端点策略、HTTP 客户端、超时、响应结构
校验和置信度阈值,但不复用 SYB 映射写入路径。Handler 只限制 64 KiB 请求体并翻译稳定
错误码;`service/purchase_spec_resolution.go` 负责请求/哈希/任务身份校验、规则优先和 AI
候选白名单;`repository/purchase_spec_resolution.go` 只保存观察和最终决策。
一次解析分成以下三个边界:
1. 短事务核对任务和领取历史,写入 `purchase_spec_resolutions.outcome=pending` 候选观察;
2. 事务外执行确定性规则或调用模型,同一进程内相同幂等身份串行,避免并发重复调用;
3. 短事务把 `pending` 完成一次,并与 `idempotency_keys` 的完整响应一起提交。
服务中断后留下的 `pending` 可以由原幂等请求接管;最终 SQL 仍只允许第一次从 `pending`
完成。响应中的 `match` 必须从保存的请求候选逐字重建。此链路不更新任务、PDD 商品和
下单执行状态;`shopee_backfill` 只保留为来源诊断信号,不参与启用或拒绝判断。
批量入口先在事务中创建 `ai_match_batches` 和逐条 `ai_match_batch_items`,再由 Admin 进程内
的有界工作池执行。工作池按业务上下文哈希归并相同明细,运行中持续写入逐条状态和汇总计数;
浏览器只轮询批次状态接口,不持有 API Key,也不承担匹配判断。Admin 启动时把遗留的