root
a6b17bae62
Server callback now rejects succeeded submissions without a non-empty answer, or whose answer contains [citation:N] markers without any sources. Desktop adapters mirror this client-side: kimi returns unknown when sources arrive without a final answer, and yuanbao returns unknown when the answer has unresolved citation markers. Yuanbao also unwraps redirect URLs, harvests citations from more DOM attributes and document URL fields, drops links pointing back at yuanbao itself, and dedupes overlapping streaming fragments. Runtime controller forwards only an explicit "succeeded" status as success; anything else becomes failed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
515 lines
16 KiB
Go
515 lines
16 KiB
Go
package app
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestDecideHeartbeatPrimaryLease(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
state heartbeatPrimaryLeaseState
|
|
want heartbeatPrimaryLeaseAction
|
|
}{
|
|
{
|
|
name: "current primary fast path",
|
|
state: heartbeatPrimaryLeaseState{
|
|
Found: true,
|
|
CurrentIsPrimary: true,
|
|
LeaseLive: true,
|
|
PrimaryAvailable: true,
|
|
},
|
|
want: heartbeatPrimaryLeaseActionPrimaryHit,
|
|
},
|
|
{
|
|
name: "current primary renews when due",
|
|
state: heartbeatPrimaryLeaseState{
|
|
Found: true,
|
|
CurrentIsPrimary: true,
|
|
LeaseLive: true,
|
|
RenewDue: true,
|
|
PrimaryAvailable: true,
|
|
},
|
|
want: heartbeatPrimaryLeaseActionRenew,
|
|
},
|
|
{
|
|
name: "non primary does not steal live lease",
|
|
state: heartbeatPrimaryLeaseState{
|
|
Found: true,
|
|
CurrentIsPrimary: false,
|
|
LeaseLive: true,
|
|
PrimaryAvailable: true,
|
|
},
|
|
want: heartbeatPrimaryLeaseActionNonPrimaryLive,
|
|
},
|
|
{
|
|
name: "expired lease can be claimed",
|
|
state: heartbeatPrimaryLeaseState{
|
|
Found: true,
|
|
CurrentIsPrimary: false,
|
|
LeaseLive: false,
|
|
PrimaryAvailable: true,
|
|
},
|
|
want: heartbeatPrimaryLeaseActionClaimExpired,
|
|
},
|
|
{
|
|
name: "revoked primary can be replaced",
|
|
state: heartbeatPrimaryLeaseState{
|
|
Found: true,
|
|
CurrentIsPrimary: false,
|
|
LeaseLive: true,
|
|
PrimaryAvailable: false,
|
|
},
|
|
want: heartbeatPrimaryLeaseActionClaimUnavailable,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got := decideHeartbeatPrimaryLease(tt.state)
|
|
if got.Action != tt.want {
|
|
t.Fatalf("decideHeartbeatPrimaryLease() = %v, want %v", got.Action, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestMonitoringHeartbeatPrimaryLeaseMetrics(t *testing.T) {
|
|
resetMonitoringHeartbeatPrimaryLeaseMetricsForTest()
|
|
t.Cleanup(resetMonitoringHeartbeatPrimaryLeaseMetricsForTest)
|
|
|
|
recordHeartbeatPrimaryLeaseMetric("primary_hit", 20*time.Millisecond)
|
|
recordHeartbeatPrimaryLeaseMetric("contention", 10*time.Millisecond)
|
|
recordHeartbeatPrimaryLeaseWriteMetric("renew")
|
|
|
|
snapshot := MonitoringHeartbeatPrimaryLeaseMetricsSnapshotValue()
|
|
if snapshot.PrimaryHitTotal != 1 {
|
|
t.Fatalf("PrimaryHitTotal = %d, want 1", snapshot.PrimaryHitTotal)
|
|
}
|
|
if snapshot.ContentionTotal != 1 {
|
|
t.Fatalf("ContentionTotal = %d, want 1", snapshot.ContentionTotal)
|
|
}
|
|
if snapshot.DBRenewTotal != 1 {
|
|
t.Fatalf("DBRenewTotal = %d, want 1", snapshot.DBRenewTotal)
|
|
}
|
|
if snapshot.DurationCount != 2 {
|
|
t.Fatalf("DurationCount = %d, want 2", snapshot.DurationCount)
|
|
}
|
|
|
|
prometheus := MonitoringHeartbeatPrimaryLeaseMetricsPrometheus()
|
|
if !strings.Contains(prometheus, `monitoring_heartbeat_primary_lease_resolutions_total{outcome="primary_hit"} 1`) {
|
|
t.Fatalf("prometheus output missing primary_hit metric: %s", prometheus)
|
|
}
|
|
if !strings.Contains(prometheus, `monitoring_heartbeat_primary_lease_db_writes_total{operation="renew"} 1`) {
|
|
t.Fatalf("prometheus output missing renew write metric: %s", prometheus)
|
|
}
|
|
}
|
|
|
|
func TestDecideMonitoringHeartbeatSnapshot(t *testing.T) {
|
|
now := time.Date(2026, 4, 24, 3, 0, 0, 0, time.UTC)
|
|
base := monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "accessible",
|
|
Connected: true,
|
|
Accessible: true,
|
|
UpdatedAt: now.Add(-time.Minute),
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
previous *monitoringPlatformAccessSnapshotState
|
|
next monitoringPlatformAccessSnapshotState
|
|
primary bool
|
|
want monitoringHeartbeatSnapshotDecision
|
|
}{
|
|
{
|
|
name: "missing snapshot inserts",
|
|
next: base,
|
|
primary: true,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
WriteSnapshot: true,
|
|
StateChanged: true,
|
|
SnapshotOperation: "insert",
|
|
},
|
|
},
|
|
{
|
|
name: "fresh unchanged accessible snapshot skips write",
|
|
previous: &base,
|
|
next: base,
|
|
primary: true,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
SnapshotOperation: "skip",
|
|
},
|
|
},
|
|
{
|
|
name: "stale unchanged snapshot refreshes",
|
|
previous: &monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "accessible",
|
|
Connected: true,
|
|
Accessible: true,
|
|
UpdatedAt: now.Add(-monitoringHeartbeatAccessSnapshotRefreshAfter - time.Second),
|
|
},
|
|
next: base,
|
|
primary: true,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
WriteSnapshot: true,
|
|
RefreshDue: true,
|
|
SnapshotOperation: "refresh",
|
|
},
|
|
},
|
|
{
|
|
name: "changed snapshot updates",
|
|
previous: &base,
|
|
next: monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "unavailable",
|
|
Connected: true,
|
|
Accessible: false,
|
|
},
|
|
primary: true,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
WriteSnapshot: true,
|
|
StateChanged: true,
|
|
PendingReconcileCandidate: true,
|
|
SnapshotOperation: "update",
|
|
},
|
|
},
|
|
{
|
|
name: "fresh unchanged primary inaccessible snapshot checks pending tasks without writing snapshot",
|
|
previous: &monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "not_logged_in",
|
|
Connected: false,
|
|
Accessible: false,
|
|
UpdatedAt: now.Add(-time.Minute),
|
|
},
|
|
next: monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "not_logged_in",
|
|
Connected: false,
|
|
Accessible: false,
|
|
},
|
|
primary: true,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
PendingReconcileCandidate: true,
|
|
SnapshotOperation: "skip",
|
|
},
|
|
},
|
|
{
|
|
name: "non-primary inaccessible snapshot does not reconcile",
|
|
previous: &monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "not_logged_in",
|
|
Connected: false,
|
|
Accessible: false,
|
|
UpdatedAt: now.Add(-time.Minute),
|
|
},
|
|
next: monitoringPlatformAccessSnapshotState{
|
|
AccessStatus: "not_logged_in",
|
|
Connected: false,
|
|
Accessible: false,
|
|
},
|
|
primary: false,
|
|
want: monitoringHeartbeatSnapshotDecision{
|
|
SnapshotOperation: "skip",
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got := decideMonitoringHeartbeatSnapshot(tt.previous, tt.next, now, tt.primary)
|
|
if got.WriteSnapshot != tt.want.WriteSnapshot ||
|
|
got.StateChanged != tt.want.StateChanged ||
|
|
got.RefreshDue != tt.want.RefreshDue ||
|
|
got.PendingReconcileCandidate != tt.want.PendingReconcileCandidate ||
|
|
got.ReconcilePending != tt.want.ReconcilePending ||
|
|
got.SnapshotOperation != tt.want.SnapshotOperation {
|
|
t.Fatalf("decideMonitoringHeartbeatSnapshot() = %+v, want %+v", got, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestMonitoringHeartbeatSnapshotMetrics(t *testing.T) {
|
|
resetMonitoringHeartbeatSnapshotMetricsForTest()
|
|
t.Cleanup(resetMonitoringHeartbeatSnapshotMetricsForTest)
|
|
|
|
recordMonitoringHeartbeatSnapshotMetrics(monitoringHeartbeatSnapshotMetricsEvent{
|
|
PlatformCount: 3,
|
|
TransactionOpened: true,
|
|
SnapshotInsertedCount: 1,
|
|
SnapshotSkippedCount: 2,
|
|
ReconciledSkippedTaskCount: 5,
|
|
ProjectionRebuildCount: 2,
|
|
Duration: 20 * time.Millisecond,
|
|
})
|
|
recordMonitoringHeartbeatSnapshotMetrics(monitoringHeartbeatSnapshotMetricsEvent{
|
|
PlatformCount: 1,
|
|
Duration: 10 * time.Millisecond,
|
|
Err: errors.New("snapshot failed"),
|
|
})
|
|
|
|
snapshot := MonitoringHeartbeatSnapshotMetricsSnapshotValue()
|
|
if snapshot.PlatformReportedTotal != 4 {
|
|
t.Fatalf("PlatformReportedTotal = %d, want 4", snapshot.PlatformReportedTotal)
|
|
}
|
|
if snapshot.TransactionOpenedTotal != 1 {
|
|
t.Fatalf("TransactionOpenedTotal = %d, want 1", snapshot.TransactionOpenedTotal)
|
|
}
|
|
if snapshot.SnapshotInsertedTotal != 1 {
|
|
t.Fatalf("SnapshotInsertedTotal = %d, want 1", snapshot.SnapshotInsertedTotal)
|
|
}
|
|
if snapshot.SnapshotSkippedTotal != 2 {
|
|
t.Fatalf("SnapshotSkippedTotal = %d, want 2", snapshot.SnapshotSkippedTotal)
|
|
}
|
|
if snapshot.ReconciledSkippedTaskTotal != 5 {
|
|
t.Fatalf("ReconciledSkippedTaskTotal = %d, want 5", snapshot.ReconciledSkippedTaskTotal)
|
|
}
|
|
if snapshot.ProjectionRebuildTotal != 2 {
|
|
t.Fatalf("ProjectionRebuildTotal = %d, want 2", snapshot.ProjectionRebuildTotal)
|
|
}
|
|
if snapshot.ErrorTotal != 1 {
|
|
t.Fatalf("ErrorTotal = %d, want 1", snapshot.ErrorTotal)
|
|
}
|
|
if snapshot.DurationCount != 2 {
|
|
t.Fatalf("DurationCount = %d, want 2", snapshot.DurationCount)
|
|
}
|
|
|
|
prometheus := MonitoringHeartbeatSnapshotMetricsPrometheus()
|
|
if !strings.Contains(prometheus, `monitoring_heartbeat_snapshot_writes_total{operation="insert"} 1`) {
|
|
t.Fatalf("prometheus output missing insert write metric: %s", prometheus)
|
|
}
|
|
if !strings.Contains(prometheus, "monitoring_heartbeat_snapshot_transactions_total 1") {
|
|
t.Fatalf("prometheus output missing transaction metric: %s", prometheus)
|
|
}
|
|
}
|
|
|
|
func TestBuildMonitoringRawPayloadKimiIncludesSearchResultsInCitationInputs(t *testing.T) {
|
|
citationTitle := "真实引用文章"
|
|
searchTitle := "搜索网页结果"
|
|
|
|
req := MonitoringTaskResultRequest{
|
|
Status: "succeeded",
|
|
Citations: []MonitoringSourceItem{
|
|
{
|
|
URL: "https://example.com/citation",
|
|
Title: &citationTitle,
|
|
},
|
|
},
|
|
SearchResults: []MonitoringSourceItem{
|
|
{
|
|
URL: "https://example.com/search",
|
|
Title: &searchTitle,
|
|
},
|
|
},
|
|
}
|
|
|
|
rawPayload, sourceInputs := buildMonitoringRawPayload("kimi", req)
|
|
if len(sourceInputs) != 2 {
|
|
t.Fatalf("buildMonitoringRawPayload() sourceInputs len = %d, want 2", len(sourceInputs))
|
|
}
|
|
seen := make(map[string]bool)
|
|
for _, item := range sourceInputs {
|
|
seen[item.URL] = true
|
|
}
|
|
if !seen["https://example.com/citation"] {
|
|
t.Fatalf("buildMonitoringRawPayload() missing citation url in source inputs: %#v", sourceInputs)
|
|
}
|
|
if !seen["https://example.com/search"] {
|
|
t.Fatalf("buildMonitoringRawPayload() missing search url in source inputs: %#v", sourceInputs)
|
|
}
|
|
|
|
var payload map[string]interface{}
|
|
if err := json.Unmarshal(rawPayload, &payload); err != nil {
|
|
t.Fatalf("json.Unmarshal(rawPayload) error = %v", err)
|
|
}
|
|
if got := len(extractMonitoringSourceItems(payload["search_results"])); got != 1 {
|
|
t.Fatalf("stored search_results len = %d, want 1", got)
|
|
}
|
|
}
|
|
|
|
func TestValidateMonitoringTaskResultSubmissionRequiresAnswerOnSuccess(t *testing.T) {
|
|
if err := validateMonitoringTaskResultSubmission("kimi", "failed", MonitoringTaskResultRequest{Status: "failed"}); err != nil {
|
|
t.Fatalf("validateMonitoringTaskResultSubmission() failed status error = %v, want nil", err)
|
|
}
|
|
|
|
answer := "有效回答"
|
|
if err := validateMonitoringTaskResultSubmission("kimi", "succeeded", MonitoringTaskResultRequest{Status: "succeeded", Answer: &answer}); err != nil {
|
|
t.Fatalf("validateMonitoringTaskResultSubmission() succeeded status with answer error = %v, want nil", err)
|
|
}
|
|
|
|
if err := validateMonitoringTaskResultSubmission("kimi", "succeeded", MonitoringTaskResultRequest{Status: "succeeded"}); err == nil {
|
|
t.Fatal("validateMonitoringTaskResultSubmission() succeeded status without answer error = nil, want error")
|
|
}
|
|
}
|
|
|
|
func TestValidateMonitoringTaskResultSubmissionRequiresSourcesForCitationMarkers(t *testing.T) {
|
|
answer := "正文 [citation:1]"
|
|
if err := validateMonitoringTaskResultSubmission("yuanbao", "succeeded", MonitoringTaskResultRequest{
|
|
Status: "succeeded",
|
|
Answer: &answer,
|
|
}); err == nil {
|
|
t.Fatal("validateMonitoringTaskResultSubmission() citation marker without sources error = nil, want error")
|
|
}
|
|
|
|
title := "引用文章"
|
|
if err := validateMonitoringTaskResultSubmission("yuanbao", "succeeded", MonitoringTaskResultRequest{
|
|
Status: "succeeded",
|
|
Answer: &answer,
|
|
Citations: []MonitoringSourceItem{
|
|
{URL: "https://example.com/source", Title: &title},
|
|
},
|
|
}); err != nil {
|
|
t.Fatalf("validateMonitoringTaskResultSubmission() citation marker with sources error = %v, want nil", err)
|
|
}
|
|
}
|
|
|
|
func TestBuildMonitoringRawPayloadQwenIncludesSearchResultsInCitationInputs(t *testing.T) {
|
|
searchTitle := "搜索网页结果"
|
|
|
|
req := MonitoringTaskResultRequest{
|
|
Status: "succeeded",
|
|
SearchResults: []MonitoringSourceItem{
|
|
{
|
|
URL: "https://example.com/search",
|
|
Title: &searchTitle,
|
|
},
|
|
},
|
|
}
|
|
|
|
_, sourceInputs := buildMonitoringRawPayload("qwen", req)
|
|
if len(sourceInputs) != 1 {
|
|
t.Fatalf("buildMonitoringRawPayload() sourceInputs len = %d, want 1", len(sourceInputs))
|
|
}
|
|
for _, item := range sourceInputs {
|
|
if item.URL != "https://example.com/search" {
|
|
t.Fatalf("buildMonitoringRawPayload() source url = %q, want search url", item.URL)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestBuildMonitoringCitationFactCanDisableContentResolution(t *testing.T) {
|
|
aliasArticleID := int64(42)
|
|
aliasPublishRecordID := int64(84)
|
|
normalizedURL := "https://example.com/article"
|
|
input := MonitoringSourceItem{
|
|
URL: "https://example.com/article?utm_source=test#section",
|
|
NormalizedURL: &normalizedURL,
|
|
}
|
|
aliasMap := map[string]monitoringAliasResolution{
|
|
normalizedURL: {
|
|
ArticleID: &aliasArticleID,
|
|
PublishRecordID: &aliasPublishRecordID,
|
|
},
|
|
}
|
|
|
|
resolvedFact, ok := buildMonitoringCitationFact(input, aliasMap, true)
|
|
if !ok {
|
|
t.Fatal("buildMonitoringCitationFact() with content resolution returned !ok")
|
|
}
|
|
if resolvedFact.ArticleID == nil || *resolvedFact.ArticleID != aliasArticleID {
|
|
t.Fatalf("resolved ArticleID = %v, want %d", resolvedFact.ArticleID, aliasArticleID)
|
|
}
|
|
if resolvedFact.PublishRecordID == nil || *resolvedFact.PublishRecordID != aliasPublishRecordID {
|
|
t.Fatalf("resolved PublishRecordID = %v, want %d", resolvedFact.PublishRecordID, aliasPublishRecordID)
|
|
}
|
|
|
|
sourceOnlyFact, ok := buildMonitoringCitationFact(input, aliasMap, false)
|
|
if !ok {
|
|
t.Fatal("buildMonitoringCitationFact() without content resolution returned !ok")
|
|
}
|
|
if sourceOnlyFact.ArticleID != nil {
|
|
t.Fatalf("source-only ArticleID = %v, want nil", sourceOnlyFact.ArticleID)
|
|
}
|
|
if sourceOnlyFact.PublishRecordID != nil {
|
|
t.Fatalf("source-only PublishRecordID = %v, want nil", sourceOnlyFact.PublishRecordID)
|
|
}
|
|
}
|
|
|
|
func TestBuildMonitoringRawPayloadDeepSeekIncludesSearchResultsInCitationInputs(t *testing.T) {
|
|
searchTitle := "搜索网页结果"
|
|
|
|
req := MonitoringTaskResultRequest{
|
|
Status: "succeeded",
|
|
SearchResults: []MonitoringSourceItem{
|
|
{
|
|
URL: "https://example.com/search",
|
|
Title: &searchTitle,
|
|
},
|
|
},
|
|
}
|
|
|
|
_, sourceInputs := buildMonitoringRawPayload("deepseek", req)
|
|
if len(sourceInputs) != 1 {
|
|
t.Fatalf("buildMonitoringRawPayload() sourceInputs len = %d, want 1", len(sourceInputs))
|
|
}
|
|
for _, item := range sourceInputs {
|
|
if item.URL != "https://example.com/search" {
|
|
t.Fatalf("buildMonitoringRawPayload() source url = %q, want search url", item.URL)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestExtractMonitoringInlineCitationsFromRawResponseKimi(t *testing.T) {
|
|
raw := []byte(`{
|
|
"inline_citation_candidates": [
|
|
{
|
|
"url": "https://example.com/a#ref",
|
|
"title": "文章 A",
|
|
"site_name": "示例站点"
|
|
},
|
|
{
|
|
"url": "https://example.com/a#other",
|
|
"title": "文章 A",
|
|
"site_name": "示例站点"
|
|
},
|
|
{
|
|
"url": "https://example.com/b",
|
|
"title": "文章 B",
|
|
"site_name": "另一个站点"
|
|
}
|
|
]
|
|
}`)
|
|
|
|
items := extractMonitoringInlineCitationsFromRawResponse(raw, "kimi")
|
|
if len(items) != 2 {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() len = %d, want 2", len(items))
|
|
}
|
|
if items[0].URL != "https://example.com/a" {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() first url = %q, want https://example.com/a", items[0].URL)
|
|
}
|
|
if items[1].URL != "https://example.com/b" {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() second url = %q, want https://example.com/b", items[1].URL)
|
|
}
|
|
}
|
|
|
|
func TestExtractMonitoringInlineCitationsFromRawResponseDeepSeekFallsBackToInlineLinks(t *testing.T) {
|
|
raw := []byte(`{
|
|
"inline_links": [
|
|
{
|
|
"url": "https://example.com/a#cite-1",
|
|
"text": "-1",
|
|
"siteName": "示例站点"
|
|
},
|
|
{
|
|
"url": "https://example.com/b",
|
|
"title": "文章 B",
|
|
"siteName": "另一个站点"
|
|
}
|
|
]
|
|
}`)
|
|
|
|
items := extractMonitoringInlineCitationsFromRawResponse(raw, "deepseek")
|
|
if len(items) != 2 {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() len = %d, want 2", len(items))
|
|
}
|
|
if items[0].URL != "https://example.com/a" {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() first url = %q, want https://example.com/a", items[0].URL)
|
|
}
|
|
if items[0].SiteName == nil || *items[0].SiteName != "示例站点" {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() first site = %v, want 示例站点", items[0].SiteName)
|
|
}
|
|
if items[1].Title == nil || *items[1].Title != "文章 B" {
|
|
t.Fatalf("extractMonitoringInlineCitationsFromRawResponse() second title = %v, want 文章 B", items[1].Title)
|
|
}
|
|
}
|