feat: integrate AI candidate review and automatic publication
This commit is contained in:
@@ -22,12 +22,12 @@ func NewCandidateRepository(pool *pgxpool.Pool) *CandidateRepository {
|
||||
// All successors of an automatically generated candidate retain the deployment binding.
|
||||
const screeningEligibility = `NOT EXISTS(SELECT 1 FROM veyra_strategy_screening x JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id JOIN veyra_strategy_version sv ON sv.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=sv.strategy_id WHERE original.candidate_id=c.id AND (s.status<>'ACTIVE' OR s.active_version_id IS DISTINCT FROM sv.id))`
|
||||
|
||||
const candidateColumns = `r.id,r.candidate_id,r.revision,r.version,r.status,r.data_mode,r.snapshot,r.author_id,r.observed_at,r.valid_until,r.published_at,r.created_at,r.updated_at,(r.revision=c.latest_revision),COALESCE(c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now() AND ` + screeningEligibility + `,false),COALESCE((SELECT reason FROM veyra_candidate_review v WHERE v.revision_id=r.id AND v.decision='WITHDRAW' ORDER BY v.input_version DESC LIMIT 1),''),(SELECT jsonb_build_object('id',x.id,'strategy_version_id',x.strategy_version_id,'strategy_name',s.name,'strategy_revision',sv.revision,'symbol',x.symbol,'exchange',x.exchange,'bar_date',x.bar_date::text,'source_id',x.source_id,'source_version',x.source_version,'source_adapter',x.source_adapter,'observed_at',x.observed_at,'received_at',x.received_at,'input_hash',x.input_hash,'status',x.status,'candidate_revision_id',x.candidate_revision_id,'created_at',x.created_at) FROM veyra_strategy_screening x JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id JOIN veyra_strategy_version sv ON sv.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=sv.strategy_id WHERE original.candidate_id=c.id LIMIT 1)`
|
||||
const candidateColumns = `r.id,r.candidate_id,r.revision,r.version,r.status,r.data_mode,r.snapshot,r.author_id,r.observed_at,r.valid_until,r.published_at,r.created_at,r.updated_at,(r.revision=c.latest_revision),COALESCE(c.published_revision_id=r.id AND r.status='PUBLISHED' AND r.valid_until>now() AND ` + screeningEligibility + `,false),COALESCE((SELECT reason FROM veyra_candidate_review v WHERE v.revision_id=r.id AND v.decision='WITHDRAW' ORDER BY v.input_version DESC LIMIT 1),''),(SELECT jsonb_build_object('id',x.id,'strategy_version_id',x.strategy_version_id,'strategy_name',s.name,'strategy_revision',sv.revision,'symbol',x.symbol,'exchange',x.exchange,'bar_date',x.bar_date::text,'source_id',x.source_id,'source_version',x.source_version,'source_adapter',x.source_adapter,'observed_at',x.observed_at,'received_at',x.received_at,'input_hash',x.input_hash,'status',x.status,'candidate_revision_id',x.candidate_revision_id,'created_at',x.created_at) FROM veyra_strategy_screening x JOIN veyra_candidate_revision original ON original.id=x.candidate_revision_id JOIN veyra_strategy_version sv ON sv.id=x.strategy_version_id JOIN veyra_strategy s ON s.id=sv.strategy_id WHERE original.candidate_id=c.id LIMIT 1),(SELECT jsonb_build_object('id',a.id,'candidate_revision_id',a.candidate_revision_id,'input_version',a.input_version,'policy_version',a.policy_version,'status',a.status,'primary_model',a.primary_model,'review_model',a.review_model,'primary_version',a.primary_version,'review_version',a.review_version,'analysis',a.analysis,'review',a.review,'message',a.message,'created_at',a.created_at,'updated_at',a.updated_at) FROM veyra_candidate_analysis a WHERE a.candidate_revision_id=r.id AND (a.input_version=r.version OR (r.status='PUBLISHED' AND a.input_version=r.version-1)) AND a.status IN ('APPROVED','PUBLISHED') ORDER BY a.updated_at DESC LIMIT 1)`
|
||||
|
||||
func scanCandidate(row pgx.Row) (model.CandidateRevision, error) {
|
||||
var r model.CandidateRevision
|
||||
var snapshot, screening []byte
|
||||
err := row.Scan(&r.ID, &r.CandidateID, &r.Revision, &r.Version, &r.Status, &r.DataMode, &snapshot, &r.AuthorID, &r.ObservedAt, &r.ValidUntil, &r.PublishedAt, &r.CreatedAt, &r.UpdatedAt, &r.IsLatest, &r.IsActivePublication, &r.WithdrawalReason, &screening)
|
||||
var snapshot, screening, analysis []byte
|
||||
err := row.Scan(&r.ID, &r.CandidateID, &r.Revision, &r.Version, &r.Status, &r.DataMode, &snapshot, &r.AuthorID, &r.ObservedAt, &r.ValidUntil, &r.PublishedAt, &r.CreatedAt, &r.UpdatedAt, &r.IsLatest, &r.IsActivePublication, &r.WithdrawalReason, &screening, &analysis)
|
||||
if err != nil {
|
||||
return r, mapError(err)
|
||||
}
|
||||
@@ -39,6 +39,11 @@ func scanCandidate(row pgx.Row) (model.CandidateRevision, error) {
|
||||
return r, apperror.ErrUnavailable
|
||||
}
|
||||
}
|
||||
if len(analysis) > 0 {
|
||||
if err := json.Unmarshal(analysis, &r.AIAnalysis); err != nil {
|
||||
return r, apperror.ErrUnavailable
|
||||
}
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
@@ -178,6 +183,14 @@ func (r *CandidateRepository) Transition(ctx context.Context, id, action string,
|
||||
return model.CandidateRevision{}, mapError(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
revision, err := transitionCandidate(ctx, tx, id, action, input, actor, requestID)
|
||||
if err != nil {
|
||||
return revision, err
|
||||
}
|
||||
return revision, mapError(tx.Commit(ctx))
|
||||
}
|
||||
|
||||
func transitionCandidate(ctx context.Context, tx pgx.Tx, id, action string, input model.CandidateActionInput, actor, requestID string) (model.CandidateRevision, error) {
|
||||
revision, err := lockCandidate(ctx, tx, id)
|
||||
if err != nil {
|
||||
return revision, err
|
||||
@@ -241,5 +254,5 @@ func (r *CandidateRepository) Transition(ctx context.Context, id, action string,
|
||||
return revision, err
|
||||
}
|
||||
}
|
||||
return revision, mapError(tx.Commit(ctx))
|
||||
return revision, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,211 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
|
||||
"git.sundynix.cn/Blizzard/Veyra/internal/model"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type CandidateAnalysisRepository struct{ pool *pgxpool.Pool }
|
||||
|
||||
func NewCandidateAnalysisRepository(pool *pgxpool.Pool) *CandidateAnalysisRepository {
|
||||
return &CandidateAnalysisRepository{pool: pool}
|
||||
}
|
||||
|
||||
const policyID = "a945c2bf-9c77-4d4a-9f4f-a986b729776c"
|
||||
|
||||
func scanPolicy(row pgx.Row) (model.AIReviewPolicy, error) {
|
||||
var p model.AIReviewPolicy
|
||||
err := row.Scan(&p.ID, &p.Enabled, &p.AutoPublish, &p.Version, &p.UpdatedBy)
|
||||
return p, mapError(err)
|
||||
}
|
||||
func (r *CandidateAnalysisRepository) Policy(ctx context.Context) (model.AIReviewPolicy, error) {
|
||||
return scanPolicy(r.pool.QueryRow(ctx, `SELECT id,enabled,auto_publish,version,updated_by FROM veyra_ai_review_policy WHERE id=$1`, policyID))
|
||||
}
|
||||
func (r *CandidateAnalysisRepository) SavePolicy(ctx context.Context, p model.AIReviewPolicy, actor, requestID string) (model.AIReviewPolicy, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return p, mapError(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
row := tx.QueryRow(ctx, `UPDATE veyra_ai_review_policy SET enabled=$2,auto_publish=$3,version=version+1,updated_by=$4,updated_at=now() WHERE id=$1 AND version=$5 RETURNING id,enabled,auto_publish,version,updated_by`, policyID, p.Enabled, p.AutoPublish, actor, p.Version)
|
||||
p, err = scanPolicy(row)
|
||||
if err == apperror.ErrNotFound {
|
||||
return p, apperror.ErrConflict
|
||||
}
|
||||
if err != nil {
|
||||
return p, err
|
||||
}
|
||||
if err = appendAudit(ctx, tx, actor, "AI_REVIEW_POLICY_SAVED", "ai_review_policy", policyID, requestID, map[string]any{"enabled": p.Enabled, "auto_publish": p.AutoPublish, "version": p.Version}); err != nil {
|
||||
return p, err
|
||||
}
|
||||
return p, mapError(tx.Commit(ctx))
|
||||
}
|
||||
func (r *CandidateAnalysisRepository) Claim(ctx context.Context, p model.AIReviewPolicy) (model.CandidateRevision, string, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return model.CandidateRevision{}, "", mapError(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('veyra:ai-claim')::bigint)`); err != nil {
|
||||
return model.CandidateRevision{}, "", mapError(err)
|
||||
}
|
||||
c, err := scanCandidate(tx.QueryRow(ctx, `SELECT `+candidateColumns+candidateJoin+` WHERE r.status='PENDING_REVIEW' AND r.revision=c.latest_revision AND r.valid_until>now() AND `+screeningEligibility+` AND NOT EXISTS(SELECT 1 FROM veyra_candidate_analysis a WHERE a.candidate_revision_id=r.id AND a.input_version=r.version AND a.policy_version=$1 AND (a.status NOT IN ('FAILED','RUNNING') OR a.attempts>=3 OR a.lease_until>now())) ORDER BY r.created_at LIMIT 1`, p.Version))
|
||||
if err != nil {
|
||||
return c, "", err
|
||||
}
|
||||
if c.Screening != nil {
|
||||
if err = tx.QueryRow(ctx, `SELECT input FROM veyra_strategy_screening WHERE id=$1`, c.Screening.ID).Scan(&c.Screening.Input); err != nil {
|
||||
return c, "", mapError(err)
|
||||
}
|
||||
}
|
||||
token := uuid.NewString()
|
||||
_, err = tx.Exec(ctx, `INSERT INTO veyra_candidate_analysis(id,candidate_revision_id,input_version,policy_version,status,claim_token,lease_until) VALUES($1,$2,$3,$4,'RUNNING',$5,now()+interval '5 minutes') ON CONFLICT(candidate_revision_id,input_version,policy_version) DO UPDATE SET status='RUNNING',attempts=veyra_candidate_analysis.attempts+1,claim_token=excluded.claim_token,lease_until=excluded.lease_until,updated_at=now(),message=''`, uuid.NewString(), c.ID, c.Version, p.Version, token)
|
||||
if err != nil {
|
||||
return c, "", mapError(err)
|
||||
}
|
||||
return c, token, mapError(tx.Commit(ctx))
|
||||
}
|
||||
func (r *CandidateAnalysisRepository) Complete(ctx context.Context, p model.AIReviewPolicy, c model.CandidateRevision, token string, primary, review model.ProviderConfig, a, b *model.AIConclusion, message string) error {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
defer tx.Rollback(context.Background())
|
||||
current, err := scanPolicy(tx.QueryRow(ctx, `SELECT id,enabled,auto_publish,version,updated_by FROM veyra_ai_review_policy WHERE id=$1 FOR SHARE`, policyID))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
status := "FAILED"
|
||||
if message == "" {
|
||||
status = "REJECTED"
|
||||
if a != nil && b != nil && a.Decision == "APPROVE" && b.Decision == "APPROVE" {
|
||||
status = "APPROVED"
|
||||
}
|
||||
}
|
||||
if !current.Enabled || current.Version != p.Version {
|
||||
status = "STALE"
|
||||
message = "审核策略已变更"
|
||||
}
|
||||
// Lock provider rows in stable key order; a changed route cannot publish old output.
|
||||
rows, err := tx.Query(ctx, `SELECT config_key,version,enabled FROM veyra_provider_config WHERE config_key IN ('market_primary','model_primary','model_review') ORDER BY config_key FOR SHARE`)
|
||||
if err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
valid := 0
|
||||
sourceValid := c.Screening == nil
|
||||
for rows.Next() {
|
||||
var key string
|
||||
var version int64
|
||||
var enabled bool
|
||||
if err = rows.Scan(&key, &version, &enabled); err != nil {
|
||||
rows.Close()
|
||||
return mapError(err)
|
||||
}
|
||||
if key == "market_primary" && c.Screening != nil && enabled && version == c.Screening.SourceVersion {
|
||||
sourceValid = true
|
||||
}
|
||||
if enabled && ((key == primary.ConfigKey && version == primary.Version) || (key == review.ConfigKey && version == review.Version)) {
|
||||
valid++
|
||||
}
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
if valid != 2 {
|
||||
status = "STALE"
|
||||
message = "模型配置已变更或未启用"
|
||||
}
|
||||
if !sourceValid {
|
||||
status = "STALE"
|
||||
message = "行情来源已停用或版本已变化"
|
||||
}
|
||||
var storedToken, storedStatus string
|
||||
if err = tx.QueryRow(ctx, `SELECT claim_token,status FROM veyra_candidate_analysis WHERE candidate_revision_id=$1 AND input_version=$2 AND policy_version=$3 FOR UPDATE`, c.ID, c.Version, p.Version).Scan(&storedToken, &storedStatus); err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
if token != storedToken || storedStatus != "RUNNING" {
|
||||
return apperror.ErrConflict
|
||||
}
|
||||
// Existing transition owns candidate/strategy locks and validates source/version/time.
|
||||
if status == "APPROVED" && current.AutoPublish && current.UpdatedBy != nil {
|
||||
var actorRole, actorStatus string
|
||||
err = tx.QueryRow(ctx, `SELECT role,status FROM veyra_user WHERE id=$1 FOR SHARE`, *current.UpdatedBy).Scan(&actorRole, &actorStatus)
|
||||
if err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
if actorRole != "ADMIN" || actorStatus != "ACTIVE" {
|
||||
status = "STALE"
|
||||
message = "自动发布授权身份已停用"
|
||||
} else {
|
||||
// Savepoint allows a rejected rule check to remain visible without publishing.
|
||||
if _, err = tx.Exec(ctx, "SAVEPOINT ai_publish"); err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
_, err = transitionCandidate(ctx, tx, c.ID, "PUBLISH", model.CandidateActionInput{ExpectedVersion: c.Version, Reason: "AI 自动审核通过；主分析与复核已保存，价格由策略规则生成"}, *current.UpdatedBy, uuid.NewString())
|
||||
if err != nil {
|
||||
if _, rollbackErr := tx.Exec(ctx, "ROLLBACK TO SAVEPOINT ai_publish"); rollbackErr != nil {
|
||||
return mapError(rollbackErr)
|
||||
}
|
||||
status = "STALE"
|
||||
message = "发布前规则校验未通过，请检查候选时效与策略版本"
|
||||
} else {
|
||||
status = "PUBLISHED"
|
||||
message = "AI 分析与复核通过，已自动发布"
|
||||
}
|
||||
}
|
||||
}
|
||||
if message == "" {
|
||||
if status == "REJECTED" {
|
||||
message = "AI 未通过，保留人工复核"
|
||||
} else {
|
||||
message = "AI 审核通过，当前为人工发布模式"
|
||||
}
|
||||
}
|
||||
rawA, _ := json.Marshal(a)
|
||||
rawB, _ := json.Marshal(b)
|
||||
_, err = tx.Exec(ctx, `UPDATE veyra_candidate_analysis SET status=$4,primary_model=$5,review_model=$6,primary_version=$7,review_version=$8,analysis=$9,review=$10,message=$11,lease_until=now()+interval '5 minutes',updated_at=now() WHERE candidate_revision_id=$1 AND input_version=$2 AND policy_version=$3`, c.ID, c.Version, p.Version, status, primary.ModelName, review.ModelName, primary.Version, review.Version, rawA, rawB, message)
|
||||
if err != nil {
|
||||
return mapError(err)
|
||||
}
|
||||
if current.UpdatedBy != nil {
|
||||
if err = appendAudit(ctx, tx, *current.UpdatedBy, "AI_CANDIDATE_COMPLETED", "candidate_revision", c.ID, uuid.NewString(), map[string]any{"status": status, "policy_version": p.Version, "input_version": c.Version}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return mapError(tx.Commit(ctx))
|
||||
}
|
||||
func (r *CandidateAnalysisRepository) List(ctx context.Context) ([]model.CandidateAnalysis, error) {
|
||||
rows, err := r.pool.Query(ctx, `SELECT id,candidate_revision_id,input_version,policy_version,status,attempts,primary_model,review_model,primary_version,review_version,analysis,review,message,created_at,updated_at FROM veyra_candidate_analysis ORDER BY updated_at DESC LIMIT 100`)
|
||||
if err != nil {
|
||||
return nil, mapError(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]model.CandidateAnalysis, 0)
|
||||
for rows.Next() {
|
||||
var v model.CandidateAnalysis
|
||||
var a, b []byte
|
||||
if err = rows.Scan(&v.ID, &v.CandidateRevisionID, &v.InputVersion, &v.PolicyVersion, &v.Status, &v.Attempts, &v.PrimaryModel, &v.ReviewModel, &v.PrimaryVersion, &v.ReviewVersion, &a, &b, &v.Message, &v.CreatedAt, &v.UpdatedAt); err != nil {
|
||||
return nil, mapError(err)
|
||||
}
|
||||
if len(a) > 0 {
|
||||
if err = json.Unmarshal(a, &v.Analysis); err != nil {
|
||||
return nil, apperror.ErrUnavailable
|
||||
}
|
||||
}
|
||||
if len(b) > 0 {
|
||||
if err = json.Unmarshal(b, &v.Review); err != nil {
|
||||
return nil, apperror.ErrUnavailable
|
||||
}
|
||||
}
|
||||
out = append(out, v)
|
||||
}
|
||||
return out, mapError(rows.Err())
|
||||
}
|
||||
@@ -99,6 +99,14 @@ func (r *IdentityRepository) ListUsers(ctx context.Context, limit int) ([]model.
|
||||
}
|
||||
|
||||
func (r *IdentityRepository) ChangeAdminCredential(ctx context.Context, previousLogin, login, passwordHash string, now time.Time) (string, error) {
|
||||
return r.changeCredential(ctx, previousLogin, login, passwordHash, now, "ADMIN", "ADMIN_CREDENTIAL_CHANGED")
|
||||
}
|
||||
|
||||
func (r *IdentityRepository) ChangeDevelopmentUserCredential(ctx context.Context, previousLogin, login, passwordHash string, now time.Time) (string, error) {
|
||||
return r.changeCredential(ctx, previousLogin, login, passwordHash, now, "USER", "DEVELOPMENT_USER_CREDENTIAL_CHANGED")
|
||||
}
|
||||
|
||||
func (r *IdentityRepository) changeCredential(ctx context.Context, previousLogin, login, passwordHash string, now time.Time, expectedRole, action string) (string, error) {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return "", mapError(err)
|
||||
@@ -109,7 +117,7 @@ func (r *IdentityRepository) ChangeAdminCredential(ctx context.Context, previous
|
||||
if err != nil {
|
||||
return "", mapError(err)
|
||||
}
|
||||
if role != "ADMIN" || status != "ACTIVE" {
|
||||
if role != expectedRole || status != "ACTIVE" {
|
||||
return "", apperror.ErrForbidden
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `UPDATE veyra_credential SET login_name=$2,password_hash=$3 WHERE user_id=$1`, userID, login, passwordHash); err != nil {
|
||||
@@ -119,7 +127,7 @@ func (r *IdentityRepository) ChangeAdminCredential(ctx context.Context, previous
|
||||
if err != nil {
|
||||
return "", mapError(err)
|
||||
}
|
||||
if err := appendAudit(ctx, tx, userID, "ADMIN_CREDENTIAL_CHANGED", "user", userID, uuid.NewString(), map[string]any{"sessions_revoked": tag.RowsAffected()}); err != nil {
|
||||
if err := appendAudit(ctx, tx, userID, action, "user", userID, uuid.NewString(), map[string]any{"sessions_revoked": tag.RowsAffected()}); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user