feat(t208): close candidate decision feedback loop
This commit is contained in:
QiuSW
@@ -383,6 +383,9 @@ func TestAuthMigrationCanRollbackWithoutRebuildingPurchaseTasks(
|
||||
if err != nil {
|
||||
t.Fatalf("migration.New() error = %v", err)
|
||||
}
|
||||
if err := runner.Down(context.Background()); err != nil {
|
||||
t.Fatalf("Down(v6) error = %v", err)
|
||||
}
|
||||
if err := runner.Down(context.Background()); err != nil {
|
||||
t.Fatalf("Down(v5) error = %v", err)
|
||||
}
|
||||
@@ -405,9 +408,9 @@ func TestAuthMigrationCanRollbackWithoutRebuildingPurchaseTasks(
|
||||
t.Fatal("purchase_tasks was lost during auth migration rollback")
|
||||
}
|
||||
if applied, err := runner.Up(context.Background()); err != nil {
|
||||
t.Fatalf("Up(v3-v4) error = %v", err)
|
||||
} else if applied != 3 {
|
||||
t.Fatalf("Up(v3-v5) applied = %d, want 3", applied)
|
||||
t.Fatalf("Up(v3-v6) error = %v", err)
|
||||
} else if applied != 4 {
|
||||
t.Fatalf("Up(v3-v6) applied = %d, want 4", applied)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,838 @@
|
||||
package sqlite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"cmroubao/backend-api/internal/domain"
|
||||
"cmroubao/backend-api/internal/usecase"
|
||||
)
|
||||
|
||||
func storeCandidateDecisionDataset(
|
||||
ctx context.Context,
|
||||
tx *sql.Tx,
|
||||
write usecase.ExecutionResultWrite,
|
||||
batch domain.ExecutionCandidateBatch,
|
||||
expired bool,
|
||||
) error {
|
||||
var candidates []usecase.ExecutionCandidate
|
||||
if err := json.Unmarshal([]byte(batch.CandidatesJSON), &candidates); err != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
var startedAt string
|
||||
var appVersion, androidVersion, pddVersion sql.NullString
|
||||
err := tx.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT execution.started_at, device.app_version,
|
||||
device.android_version, device.pdd_version
|
||||
FROM task_executions AS execution
|
||||
JOIN devices AS device ON device.id = execution.device_id
|
||||
WHERE execution.id = ? AND execution.task_id = ?`,
|
||||
write.ExecutionID,
|
||||
write.TaskID,
|
||||
).Scan(&startedAt, &appVersion, &androidVersion, &pddVersion)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_search_runs (
|
||||
execution_id, task_id, task_content_sha256, execution_mode,
|
||||
search_query, app_version, android_version, pdd_version,
|
||||
started_at, received_at, observation_count, collection_complete,
|
||||
received_after_execution_expiry
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)`,
|
||||
write.ExecutionID,
|
||||
write.TaskID,
|
||||
batch.TaskContentSHA256,
|
||||
batch.ExecutionMode,
|
||||
batch.SearchQuery,
|
||||
nullableSQLString(appVersion),
|
||||
nullableSQLString(androidVersion),
|
||||
nullableSQLString(pddVersion),
|
||||
startedAt,
|
||||
formatTimestamp(write.Now),
|
||||
len(candidates),
|
||||
expired,
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
for _, candidate := range candidates {
|
||||
evidenceJSON, marshalErr := json.Marshal(candidate.EvidenceAssetIDs)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
collectionStatus := "PARTIAL"
|
||||
if len(candidate.EvidenceAssetIDs) > 0 {
|
||||
collectionStatus = "COMPLETE"
|
||||
}
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_observations (
|
||||
execution_id, task_id, ordinal, title, sku_text, price_text,
|
||||
product_url, image_url, evidence_asset_ids_json,
|
||||
collection_status, observed_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
write.ExecutionID,
|
||||
write.TaskID,
|
||||
candidate.Ordinal,
|
||||
candidate.Title,
|
||||
candidate.SKUText,
|
||||
candidate.Price,
|
||||
candidate.ProductURL,
|
||||
candidate.ImageURL,
|
||||
string(evidenceJSON),
|
||||
collectionStatus,
|
||||
formatTimestamp(write.Now),
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
}
|
||||
if batch.ProvenanceJSON != nil {
|
||||
var provenance usecase.ExecutionProvenance
|
||||
if err := json.Unmarshal(
|
||||
[]byte(*batch.ProvenanceJSON),
|
||||
&provenance,
|
||||
); err != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
resultHash := hashCandidateResult(
|
||||
batch.CandidatesJSON,
|
||||
batch.RecommendationJSON,
|
||||
)
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO model_runs (
|
||||
execution_id, provider_id, model, prompt_version, schema_version,
|
||||
recommendation_threshold, request_sha256, result_sha256,
|
||||
created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
write.ExecutionID,
|
||||
provenance.ProviderID,
|
||||
provenance.Model,
|
||||
provenance.PromptVersion,
|
||||
provenance.SchemaVersion,
|
||||
0.75,
|
||||
write.RequestHash,
|
||||
resultHash,
|
||||
formatTimestamp(write.Now),
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
for _, candidate := range candidates {
|
||||
if candidate.Evaluation == nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
evaluation := candidate.Evaluation
|
||||
matchedJSON, marshalErr := json.Marshal(evaluation.Matched)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
missingJSON, marshalErr := json.Marshal(
|
||||
evaluation.MissingOrUncertain,
|
||||
)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
rejectionsJSON, marshalErr := json.Marshal(
|
||||
evaluation.RejectionReasons,
|
||||
)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
constraintsJSON, marshalErr := json.Marshal(
|
||||
evaluation.HardConstraints,
|
||||
)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_evaluations (
|
||||
execution_id, candidate_ordinal, decision, score,
|
||||
confidence, matched_json, missing_or_uncertain_json,
|
||||
rejection_reasons_json, hard_constraints_json, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
write.ExecutionID,
|
||||
candidate.Ordinal,
|
||||
evaluation.Decision,
|
||||
evaluation.Score,
|
||||
evaluation.Confidence,
|
||||
string(matchedJSON),
|
||||
string(missingJSON),
|
||||
string(rejectionsJSON),
|
||||
string(constraintsJSON),
|
||||
formatTimestamp(write.Now),
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if batch.RecommendationJSON != nil {
|
||||
var recommendation usecase.CandidateRecommendation
|
||||
if err := json.Unmarshal(
|
||||
[]byte(*batch.RecommendationJSON),
|
||||
&recommendation,
|
||||
); err != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
reasonsJSON, marshalErr := json.Marshal(recommendation.Reasons)
|
||||
if marshalErr != nil {
|
||||
return usecase.ErrRepositoryInvariant
|
||||
}
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_recommendations (
|
||||
execution_id, candidate_ordinal, conclusion, policy_version,
|
||||
reasons_json, created_at
|
||||
) VALUES (?, ?, 'SUGGESTED', ?, ?, ?)`,
|
||||
write.ExecutionID,
|
||||
recommendation.CandidateOrdinal,
|
||||
recommendation.PolicyVersion,
|
||||
string(reasonsJSON),
|
||||
formatTimestamp(write.Now),
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) StoreCandidateHumanReview(
|
||||
ctx context.Context,
|
||||
write usecase.ExecutionResultWrite,
|
||||
candidate domain.CandidateHumanReview,
|
||||
) (domain.CandidateHumanReview, bool, error) {
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
record, found, err := lookupExecutionResultRequest(ctx, tx, write)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
if found {
|
||||
if err := validateExecutionResultReplay(record, write); err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
if record.ResourceID == nil {
|
||||
return domain.CandidateHumanReview{}, false, usecase.ErrRepositoryInvariant
|
||||
}
|
||||
review, err := getCandidateHumanReview(ctx, tx, *record.ResourceID)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
return review, true, nil
|
||||
}
|
||||
task, _, expired, err := authorizeExecutionResult(ctx, tx, write)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
if usecase.TaskContentSHA256(task) != candidate.TaskContentSHA256 {
|
||||
return domain.CandidateHumanReview{}, false, usecase.ErrTaskVersionConflict
|
||||
}
|
||||
if err := validateHumanReviewAgainstDataset(
|
||||
ctx,
|
||||
tx,
|
||||
task,
|
||||
candidate,
|
||||
); err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
latestID, latestVersion, err := latestCandidateHumanReview(
|
||||
ctx,
|
||||
tx,
|
||||
write.ExecutionID,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
switch {
|
||||
case latestID == nil && candidate.SupersedesReviewID != nil:
|
||||
return domain.CandidateHumanReview{}, false, usecase.ErrTaskStateConflict
|
||||
case latestID != nil && (candidate.SupersedesReviewID == nil ||
|
||||
*candidate.SupersedesReviewID != *latestID):
|
||||
return domain.CandidateHumanReview{}, false, usecase.ErrTaskStateConflict
|
||||
}
|
||||
candidate.Version = latestVersion + 1
|
||||
candidate.ReceivedAfterExecutionExpiry = expired
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_human_reviews (
|
||||
id, execution_id, task_id, version, reason_schema_version, outcome,
|
||||
selected_candidate_ordinal, primary_reason_code, note,
|
||||
supersedes_review_id, actor_user_id, actor_device_id, created_at,
|
||||
received_after_execution_expiry
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
candidate.ID,
|
||||
candidate.ExecutionID,
|
||||
candidate.TaskID,
|
||||
candidate.Version,
|
||||
candidate.ReasonSchemaVersion,
|
||||
candidate.Outcome,
|
||||
nullableInt(candidate.SelectedCandidateOrdinal),
|
||||
candidate.PrimaryReasonCode,
|
||||
candidate.Note,
|
||||
nullableString(candidate.SupersedesReviewID),
|
||||
candidate.ActorUserID,
|
||||
nullableString(candidate.ActorDeviceID),
|
||||
formatTimestamp(candidate.CreatedAt),
|
||||
expired,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
for _, item := range candidate.Items {
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_human_review_items (
|
||||
review_id, candidate_ordinal, label, primary_reason_code, note
|
||||
) VALUES (?, ?, ?, ?, ?)`,
|
||||
candidate.ID,
|
||||
item.CandidateOrdinal,
|
||||
item.Label,
|
||||
item.PrimaryReasonCode,
|
||||
item.Note,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
for _, reason := range item.ReasonCodes {
|
||||
_, err = tx.ExecContext(
|
||||
ctx,
|
||||
`INSERT INTO candidate_human_review_reasons (
|
||||
review_id, candidate_ordinal, reason_code
|
||||
) VALUES (?, ?, ?)`,
|
||||
candidate.ID,
|
||||
item.CandidateOrdinal,
|
||||
reason,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := insertExecutionResultRequest(
|
||||
ctx,
|
||||
tx,
|
||||
write,
|
||||
&candidate.ID,
|
||||
); err != nil {
|
||||
return domain.CandidateHumanReview{}, false, err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return domain.CandidateHumanReview{}, false, repositoryFailure(err)
|
||||
}
|
||||
return candidate, false, nil
|
||||
}
|
||||
|
||||
func validateHumanReviewAgainstDataset(
|
||||
ctx context.Context,
|
||||
tx *sql.Tx,
|
||||
task domain.PurchaseTask,
|
||||
review domain.CandidateHumanReview,
|
||||
) error {
|
||||
rows, err := tx.QueryContext(
|
||||
ctx,
|
||||
`SELECT ordinal FROM candidate_observations
|
||||
WHERE execution_id = ? AND task_id = ?
|
||||
ORDER BY ordinal ASC`,
|
||||
review.ExecutionID,
|
||||
review.TaskID,
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
ordinals := make([]int, 0, 5)
|
||||
for rows.Next() {
|
||||
var ordinal int
|
||||
if err := rows.Scan(&ordinal); err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
ordinals = append(ordinals, ordinal)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
if len(review.Items) != len(ordinals) {
|
||||
return usecase.ErrTaskStateConflict
|
||||
}
|
||||
expected := make(map[int]struct{}, len(ordinals))
|
||||
for _, ordinal := range ordinals {
|
||||
expected[ordinal] = struct{}{}
|
||||
}
|
||||
for _, item := range review.Items {
|
||||
if _, found := expected[item.CandidateOrdinal]; !found {
|
||||
return usecase.ErrTaskStateConflict
|
||||
}
|
||||
for _, reason := range item.ReasonCodes {
|
||||
if task.MaxBudgetCents == nil &&
|
||||
(reason == "PRICE_ACCEPTABLE" || reason == "PRICE_TOO_HIGH") {
|
||||
return usecase.ErrTaskStateConflict
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func latestCandidateHumanReview(
|
||||
ctx context.Context,
|
||||
queryer queryRower,
|
||||
executionID string,
|
||||
) (*string, int, error) {
|
||||
var id string
|
||||
var version int
|
||||
err := queryer.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT id, version FROM candidate_human_reviews
|
||||
WHERE execution_id = ?
|
||||
ORDER BY version DESC
|
||||
LIMIT 1`,
|
||||
executionID,
|
||||
).Scan(&id, &version)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, 0, repositoryFailure(err)
|
||||
}
|
||||
return &id, version, nil
|
||||
}
|
||||
|
||||
func getCandidateHumanReview(
|
||||
ctx context.Context,
|
||||
queryer queryer,
|
||||
reviewID string,
|
||||
) (domain.CandidateHumanReview, error) {
|
||||
var review domain.CandidateHumanReview
|
||||
var selected sql.NullInt64
|
||||
var supersedes, device sql.NullString
|
||||
var createdAt string
|
||||
err := queryer.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT review.id, review.task_id, review.execution_id,
|
||||
run.task_content_sha256, review.version,
|
||||
review.reason_schema_version, review.outcome,
|
||||
review.selected_candidate_ordinal, review.primary_reason_code,
|
||||
review.note, review.supersedes_review_id, review.actor_user_id,
|
||||
review.actor_device_id, review.created_at,
|
||||
review.received_after_execution_expiry
|
||||
FROM candidate_human_reviews AS review
|
||||
JOIN candidate_search_runs AS run
|
||||
ON run.execution_id = review.execution_id
|
||||
WHERE review.id = ?`,
|
||||
reviewID,
|
||||
).Scan(
|
||||
&review.ID,
|
||||
&review.TaskID,
|
||||
&review.ExecutionID,
|
||||
&review.TaskContentSHA256,
|
||||
&review.Version,
|
||||
&review.ReasonSchemaVersion,
|
||||
&review.Outcome,
|
||||
&selected,
|
||||
&review.PrimaryReasonCode,
|
||||
&review.Note,
|
||||
&supersedes,
|
||||
&review.ActorUserID,
|
||||
&device,
|
||||
&createdAt,
|
||||
&review.ReceivedAfterExecutionExpiry,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return domain.CandidateHumanReview{}, usecase.ErrRepositoryInvariant
|
||||
}
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, repositoryFailure(err)
|
||||
}
|
||||
if selected.Valid {
|
||||
value := int(selected.Int64)
|
||||
review.SelectedCandidateOrdinal = &value
|
||||
}
|
||||
review.SupersedesReviewID = nullableStringFromSQL(supersedes)
|
||||
review.ActorDeviceID = nullableStringFromSQL(device)
|
||||
review.CreatedAt, err = parseTimestamp(createdAt)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, repositoryFailure(err)
|
||||
}
|
||||
rows, err := queryer.QueryContext(
|
||||
ctx,
|
||||
`SELECT candidate_ordinal, label, primary_reason_code, note
|
||||
FROM candidate_human_review_items
|
||||
WHERE review_id = ?
|
||||
ORDER BY candidate_ordinal ASC`,
|
||||
reviewID,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, repositoryFailure(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var item domain.CandidateHumanReviewItem
|
||||
if err := rows.Scan(
|
||||
&item.CandidateOrdinal,
|
||||
&item.Label,
|
||||
&item.PrimaryReasonCode,
|
||||
&item.Note,
|
||||
); err != nil {
|
||||
return domain.CandidateHumanReview{}, repositoryFailure(err)
|
||||
}
|
||||
reasons, err := candidateHumanReviewReasons(
|
||||
ctx,
|
||||
queryer,
|
||||
reviewID,
|
||||
item.CandidateOrdinal,
|
||||
)
|
||||
if err != nil {
|
||||
return domain.CandidateHumanReview{}, err
|
||||
}
|
||||
item.ReasonCodes = reasons
|
||||
review.Items = append(review.Items, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return domain.CandidateHumanReview{}, repositoryFailure(err)
|
||||
}
|
||||
return review, nil
|
||||
}
|
||||
|
||||
func getCandidateDecisionDataset(
|
||||
ctx context.Context,
|
||||
queryer queryer,
|
||||
taskID string,
|
||||
executionID string,
|
||||
) (*domain.CandidateDecisionDataset, error) {
|
||||
var run domain.CandidateSearchRun
|
||||
var appVersion, androidVersion, pddVersion sql.NullString
|
||||
var startedAt, receivedAt string
|
||||
err := queryer.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT task_id, execution_id, task_content_sha256, execution_mode,
|
||||
search_query, app_version, android_version, pdd_version,
|
||||
started_at, received_at, observation_count, collection_complete,
|
||||
received_after_execution_expiry
|
||||
FROM candidate_search_runs
|
||||
WHERE task_id = ? AND execution_id = ?`,
|
||||
taskID,
|
||||
executionID,
|
||||
).Scan(
|
||||
&run.TaskID,
|
||||
&run.ExecutionID,
|
||||
&run.TaskContentSHA256,
|
||||
&run.ExecutionMode,
|
||||
&run.SearchQuery,
|
||||
&appVersion,
|
||||
&androidVersion,
|
||||
&pddVersion,
|
||||
&startedAt,
|
||||
&receivedAt,
|
||||
&run.ObservationCount,
|
||||
&run.CollectionComplete,
|
||||
&run.ReceivedAfterExecutionExpiry,
|
||||
)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
run.AppVersion = nullableStringFromSQL(appVersion)
|
||||
run.AndroidVersion = nullableStringFromSQL(androidVersion)
|
||||
run.PDDVersion = nullableStringFromSQL(pddVersion)
|
||||
run.StartedAt, err = parseTimestamp(startedAt)
|
||||
if err == nil {
|
||||
run.ReceivedAt, err = parseTimestamp(receivedAt)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
dataset := &domain.CandidateDecisionDataset{
|
||||
SearchRun: &run,
|
||||
Observations: make([]domain.CandidateObservation, 0, run.ObservationCount),
|
||||
Evaluations: make([]domain.CandidateEvaluationRecord, 0, run.ObservationCount),
|
||||
HumanReviews: make([]domain.CandidateHumanReview, 0),
|
||||
}
|
||||
rows, err := queryer.QueryContext(
|
||||
ctx,
|
||||
`SELECT task_id, execution_id, ordinal, title, sku_text, price_text,
|
||||
product_url, image_url, evidence_asset_ids_json,
|
||||
collection_status, observed_at
|
||||
FROM candidate_observations
|
||||
WHERE task_id = ? AND execution_id = ?
|
||||
ORDER BY ordinal ASC`,
|
||||
taskID,
|
||||
executionID,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
for rows.Next() {
|
||||
var observation domain.CandidateObservation
|
||||
var evidenceJSON, observedAt string
|
||||
if err := rows.Scan(
|
||||
&observation.TaskID,
|
||||
&observation.ExecutionID,
|
||||
&observation.Ordinal,
|
||||
&observation.Title,
|
||||
&observation.SKUText,
|
||||
&observation.PriceText,
|
||||
&observation.ProductURL,
|
||||
&observation.ImageURL,
|
||||
&evidenceJSON,
|
||||
&observation.CollectionStatus,
|
||||
&observedAt,
|
||||
); err != nil {
|
||||
_ = rows.Close()
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
if err := json.Unmarshal(
|
||||
[]byte(evidenceJSON),
|
||||
&observation.EvidenceAssetIDs,
|
||||
); err != nil {
|
||||
_ = rows.Close()
|
||||
return nil, usecase.ErrRepositoryInvariant
|
||||
}
|
||||
observation.ObservedAt, err = parseTimestamp(observedAt)
|
||||
if err != nil {
|
||||
_ = rows.Close()
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
dataset.Observations = append(dataset.Observations, observation)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
if err := loadCandidateModelData(ctx, queryer, executionID, dataset); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := loadCandidateHumanReviews(ctx, queryer, executionID, dataset); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return dataset, nil
|
||||
}
|
||||
|
||||
func loadCandidateModelData(
|
||||
ctx context.Context,
|
||||
queryer queryer,
|
||||
executionID string,
|
||||
dataset *domain.CandidateDecisionDataset,
|
||||
) error {
|
||||
var model domain.CandidateModelRun
|
||||
var createdAt string
|
||||
err := queryer.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT execution_id, provider_id, model, prompt_version,
|
||||
schema_version, recommendation_threshold, request_sha256,
|
||||
result_sha256, created_at
|
||||
FROM model_runs
|
||||
WHERE execution_id = ?`,
|
||||
executionID,
|
||||
).Scan(
|
||||
&model.ExecutionID,
|
||||
&model.ProviderID,
|
||||
&model.Model,
|
||||
&model.PromptVersion,
|
||||
&model.SchemaVersion,
|
||||
&model.RecommendationThreshold,
|
||||
&model.RequestSHA256,
|
||||
&model.ResultSHA256,
|
||||
&createdAt,
|
||||
)
|
||||
if err == nil {
|
||||
model.CreatedAt, err = parseTimestamp(createdAt)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
dataset.ModelRun = &model
|
||||
} else if !errors.Is(err, sql.ErrNoRows) {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
rows, err := queryer.QueryContext(
|
||||
ctx,
|
||||
`SELECT execution_id, candidate_ordinal, decision, score,
|
||||
confidence, matched_json, missing_or_uncertain_json,
|
||||
rejection_reasons_json, hard_constraints_json, created_at
|
||||
FROM candidate_evaluations
|
||||
WHERE execution_id = ?
|
||||
ORDER BY candidate_ordinal ASC`,
|
||||
executionID,
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
for rows.Next() {
|
||||
var evaluation domain.CandidateEvaluationRecord
|
||||
var evaluationAt string
|
||||
if err := rows.Scan(
|
||||
&evaluation.ExecutionID,
|
||||
&evaluation.CandidateOrdinal,
|
||||
&evaluation.Decision,
|
||||
&evaluation.Score,
|
||||
&evaluation.Confidence,
|
||||
&evaluation.MatchedJSON,
|
||||
&evaluation.MissingOrUncertainJSON,
|
||||
&evaluation.RejectionReasonsJSON,
|
||||
&evaluation.HardConstraintsJSON,
|
||||
&evaluationAt,
|
||||
); err != nil {
|
||||
_ = rows.Close()
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
evaluation.CreatedAt, err = parseTimestamp(evaluationAt)
|
||||
if err != nil {
|
||||
_ = rows.Close()
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
dataset.Evaluations = append(dataset.Evaluations, evaluation)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
var recommendation domain.CandidateRecommendationRecord
|
||||
var reasonsJSON, recommendationAt string
|
||||
err = queryer.QueryRowContext(
|
||||
ctx,
|
||||
`SELECT execution_id, candidate_ordinal, conclusion, policy_version,
|
||||
reasons_json, created_at
|
||||
FROM candidate_recommendations
|
||||
WHERE execution_id = ?`,
|
||||
executionID,
|
||||
).Scan(
|
||||
&recommendation.ExecutionID,
|
||||
&recommendation.CandidateOrdinal,
|
||||
&recommendation.Conclusion,
|
||||
&recommendation.PolicyVersion,
|
||||
&reasonsJSON,
|
||||
&recommendationAt,
|
||||
)
|
||||
if err == nil {
|
||||
recommendation.ReasonsJSON = reasonsJSON
|
||||
recommendation.CreatedAt, err = parseTimestamp(recommendationAt)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
dataset.Recommendation = &recommendation
|
||||
} else if !errors.Is(err, sql.ErrNoRows) {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadCandidateHumanReviews(
|
||||
ctx context.Context,
|
||||
queryer queryer,
|
||||
executionID string,
|
||||
dataset *domain.CandidateDecisionDataset,
|
||||
) error {
|
||||
rows, err := queryer.QueryContext(
|
||||
ctx,
|
||||
`SELECT id FROM candidate_human_reviews
|
||||
WHERE execution_id = ?
|
||||
ORDER BY version ASC`,
|
||||
executionID,
|
||||
)
|
||||
if err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
reviewIDs := make([]string, 0)
|
||||
for rows.Next() {
|
||||
var reviewID string
|
||||
if err := rows.Scan(&reviewID); err != nil {
|
||||
_ = rows.Close()
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
reviewIDs = append(reviewIDs, reviewID)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return repositoryFailure(err)
|
||||
}
|
||||
for _, reviewID := range reviewIDs {
|
||||
review, err := getCandidateHumanReview(ctx, queryer, reviewID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dataset.HumanReviews = append(dataset.HumanReviews, review)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func candidateHumanReviewReasons(
|
||||
ctx context.Context,
|
||||
queryer queryer,
|
||||
reviewID string,
|
||||
ordinal int,
|
||||
) ([]string, error) {
|
||||
rows, err := queryer.QueryContext(
|
||||
ctx,
|
||||
`SELECT reason_code FROM candidate_human_review_reasons
|
||||
WHERE review_id = ? AND candidate_ordinal = ?
|
||||
ORDER BY reason_code ASC`,
|
||||
reviewID,
|
||||
ordinal,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
reasons := make([]string, 0)
|
||||
for rows.Next() {
|
||||
var reason string
|
||||
if err := rows.Scan(&reason); err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
reasons = append(reasons, reason)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
return reasons, nil
|
||||
}
|
||||
|
||||
func hashCandidateResult(candidates string, recommendation *string) string {
|
||||
digest := sha256.New()
|
||||
_, _ = digest.Write([]byte(candidates))
|
||||
_, _ = digest.Write([]byte{0})
|
||||
if recommendation != nil {
|
||||
_, _ = digest.Write([]byte(*recommendation))
|
||||
}
|
||||
return hex.EncodeToString(digest.Sum(nil))
|
||||
}
|
||||
|
||||
func nullableSQLString(value sql.NullString) any {
|
||||
if !value.Valid {
|
||||
return nil
|
||||
}
|
||||
return value.String
|
||||
}
|
||||
|
||||
func nullableInt(value *int) any {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
return *value
|
||||
}
|
||||
@@ -183,6 +183,15 @@ func (s *Store) StoreExecutionCandidates(
|
||||
if err != nil {
|
||||
return false, repositoryFailure(err)
|
||||
}
|
||||
if err := storeCandidateDecisionDataset(
|
||||
ctx,
|
||||
tx,
|
||||
write,
|
||||
candidate,
|
||||
expired,
|
||||
); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if err := insertExecutionResultRequest(ctx, tx, write, nil); err != nil {
|
||||
return false, err
|
||||
}
|
||||
@@ -825,6 +834,15 @@ func getExecutionReport(
|
||||
} else if !errors.Is(err, sql.ErrNoRows) {
|
||||
return nil, repositoryFailure(err)
|
||||
}
|
||||
report.DecisionDataset, err = getCandidateDecisionDataset(
|
||||
ctx,
|
||||
queryer,
|
||||
taskID,
|
||||
executionID,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var outcome domain.ExecutionOutcome
|
||||
var mode, hash, result, reason, selected, evidenceIDs, code, message, step sql.NullString
|
||||
var retryable sql.NullBool
|
||||
|
||||
Reference in New Issue
Block a user