diff --git a/db/migrations/000004_delivery_audit.down.sql b/db/migrations/000004_delivery_audit.down.sql new file mode 100644 index 0000000..39e17a2 --- /dev/null +++ b/db/migrations/000004_delivery_audit.down.sql @@ -0,0 +1,16 @@ +-- 000004_delivery_audit down migration +-- Drop in reverse order: index first, then columns. + +DROP INDEX IF EXISTS idx_messages_tenant_status_updated; + +ALTER TABLE messages + DROP COLUMN IF EXISTS failure_code; + +ALTER TABLE messages + DROP COLUMN IF EXISTS provider_message_id; + +ALTER TABLE messages + DROP COLUMN IF EXISTS delivery_method; + +ALTER TABLE messages + DROP COLUMN IF EXISTS delivered_at; \ No newline at end of file diff --git a/db/migrations/000004_delivery_audit.up.sql b/db/migrations/000004_delivery_audit.up.sql new file mode 100644 index 0000000..64778bd --- /dev/null +++ b/db/migrations/000004_delivery_audit.up.sql @@ -0,0 +1,20 @@ +-- 000004_delivery_audit: delivery audit columns on messages +-- Adds delivered_at, delivery_method, provider_message_id, failure_code +-- with a tenant+status+updated_at query index. + +ALTER TABLE messages + ADD COLUMN IF NOT EXISTS delivered_at TIMESTAMPTZ; + +ALTER TABLE messages + ADD COLUMN IF NOT EXISTS delivery_method TEXT NOT NULL DEFAULT '' + CHECK (delivery_method IN ('', 'card', 'markdown')); + +ALTER TABLE messages + ADD COLUMN IF NOT EXISTS provider_message_id TEXT NOT NULL DEFAULT ''; + +ALTER TABLE messages + ADD COLUMN IF NOT EXISTS failure_code TEXT NOT NULL DEFAULT '' + CHECK (failure_code IN ('', 'channel_delivery_failed')); + +CREATE INDEX IF NOT EXISTS idx_messages_tenant_status_updated + ON messages (tenant_key, status, updated_at DESC); \ No newline at end of file diff --git a/db/migrations/000004_delivery_audit_contract_test.go b/db/migrations/000004_delivery_audit_contract_test.go new file mode 100644 index 0000000..e67abc9 --- /dev/null +++ b/db/migrations/000004_delivery_audit_contract_test.go @@ -0,0 +1,200 @@ +// Package migrations_test contains source-level contract tests for the +// 000004_delivery_audit migration. No real database connection is used. +package migrations_test + +import ( + "os" + "path/filepath" + "regexp" + "strings" + "testing" +) + +func read004UpMigration(t *testing.T) string { + t.Helper() + data, err := os.ReadFile(filepath.Join(".", "000004_delivery_audit.up.sql")) + if err != nil { + t.Fatalf("read up migration: %v", err) + } + return string(data) +} + +func read004DownMigration(t *testing.T) string { + t.Helper() + data, err := os.ReadFile(filepath.Join(".", "000004_delivery_audit.down.sql")) + if err != nil { + t.Fatalf("read down migration: %v", err) + } + return string(data) +} + +// TestUp004UsesIfNotExists ensures ALTER TABLE uses IF NOT EXISTS for new columns. +func TestUp004UsesIfNotExists(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + if !strings.Contains(upper, "ADD COLUMN IF NOT EXISTS") { + t.Error("up migration must use ADD COLUMN IF NOT EXISTS for idempotency") + } + if !strings.Contains(upper, "CREATE INDEX IF NOT EXISTS") { + t.Error("up migration must use CREATE INDEX IF NOT EXISTS for idempotency") + } + + // At least 4 new columns: delivered_at, delivery_method, provider_message_id, failure_code + if got := countMatches(up, `(?i)ADD\s+COLUMN\s+IF\s+NOT\s+EXISTS`); got < 4 { + t.Errorf("expected at least 4 ADD COLUMN IF NOT EXISTS, got %d", got) + } + // At least 1 new index + if got := countMatches(up, `(?i)CREATE\s+INDEX\s+IF\s+NOT\s+EXISTS`); got < 1 { + t.Errorf("expected at least 1 CREATE INDEX IF NOT EXISTS, got %d", got) + } +} + +// TestUp004AddsDeliveryColumns ensures all 4 delivery audit columns are added to messages. +func TestUp004AddsDeliveryColumns(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + columns := []string{ + "DELIVERED_AT TIMESTAMPTZ", + "DELIVERY_METHOD TEXT NOT NULL DEFAULT ''", + "PROVIDER_MESSAGE_ID TEXT NOT NULL DEFAULT ''", + "FAILURE_CODE TEXT NOT NULL DEFAULT ''", + } + for _, col := range columns { + if !strings.Contains(upper, col) { + t.Errorf("up migration must add column %q", col) + } + } +} + +// TestUp004DeliveryMethodCheck ensures delivery_method CHECK constraint. +func TestUp004DeliveryMethodCheck(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + // delivery_method must have CHECK constraint + if !strings.Contains(upper, "DELIVERY_METHOD") { + t.Fatal("up migration must add delivery_method column") + } + // CHECK (delivery_method IN ('', 'card', 'markdown')) + if !strings.Contains(upper, "CARD") || !strings.Contains(upper, "MARKDOWN") { + t.Error("delivery_method CHECK must include 'card' and 'markdown'") + } +} + +func TestUp004FailureCodeCheck(t *testing.T) { + up := strings.ToUpper(read004UpMigration(t)) + if !strings.Contains(up, "CHECK (FAILURE_CODE IN ('', 'CHANNEL_DELIVERY_FAILED'))") { + t.Error("failure_code must be constrained to safe enumerated values") + } +} + +// TestUp004AddsTenantStatusUpdatedAtIndex ensures the query index is created. +func TestUp004AddsTenantStatusUpdatedAtIndex(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + // Must have an index on (tenant_key, status, updated_at DESC) + hasDesired := strings.Contains(upper, "TENANT_KEY") && + strings.Contains(upper, "STATUS") && + strings.Contains(upper, "UPDATED_AT") && + strings.Contains(upper, "DESC") + if !hasDesired { + t.Error("up migration must create index on (tenant_key, status, updated_at DESC)") + } +} + +// TestUp004NoForbiddenColumns ensures no secrets/protocol columns are introduced. +func TestUp004NoForbiddenColumns(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + forbidden := []string{ + "RAW_PAYLOAD", "SESSIONWEBHOOK", "DOWNLOAD_CODE", + "DATAURI", "DATA_URI", "TOKEN", "RAW_RESPONSE", "RAW_ERROR", + "PROVIDER_RESPONSE", "API_RESPONSE", + } + for _, col := range forbidden { + if strings.Contains(upper, col) { + t.Errorf("migration must not contain forbidden column %q", col) + } + } +} + +// TestUp004DoesNotModifyOtherTables ensures migration only touches messages. +func TestUp004DoesNotModifyOtherTables(t *testing.T) { + up := read004UpMigration(t) + upper := strings.ToUpper(up) + + // Must not create/drop turns, session_mappings, channel_actors, channel_conversations + if strings.Contains(upper, "CREATE TABLE") { + t.Error("migration 004 must not CREATE TABLE") + } + if strings.Contains(upper, "DROP TABLE") { + t.Error("migration 004 must not DROP TABLE") + } + if strings.Contains(upper, "ALTER TABLE TURNS") { + t.Error("migration 004 must not ALTER TABLE turns") + } + if strings.Contains(upper, "ALTER TABLE SESSION_MAPPINGS") { + t.Error("migration 004 must not ALTER TABLE session_mappings") + } + if strings.Contains(upper, "ALTER TABLE CHANNEL_ACTORS") { + t.Error("migration 004 must not ALTER TABLE channel_actors") + } + if strings.Contains(upper, "ALTER TABLE CHANNEL_CONVERSATIONS") { + t.Error("migration 004 must not ALTER TABLE channel_conversations") + } +} + +// TestDown004DropsDeliveryColumns ensures down drops columns in reverse order. +func TestDown004DropsDeliveryColumns(t *testing.T) { + down := read004DownMigration(t) + upper := strings.ToUpper(down) + + columns := []string{ + "FAILURE_CODE", + "PROVIDER_MESSAGE_ID", + "DELIVERY_METHOD", + "DELIVERED_AT", + } + for _, col := range columns { + if !strings.Contains(upper, "DROP COLUMN IF EXISTS "+col) { + t.Errorf("down migration must drop %s with IF EXISTS", col) + } + } + + // Must drop the index too + if !strings.Contains(upper, "DROP INDEX IF EXISTS") { + t.Error("down migration must drop the delivery audit index") + } +} + +// TestDown004DoesNotDropBaselineTable ensures down does not drop messages table. +func TestDown004DoesNotDropBaselineTable(t *testing.T) { + down := read004DownMigration(t) + upper := strings.ToUpper(down) + + if strings.Contains(upper, "DROP TABLE IF EXISTS MESSAGES") { + t.Error("down migration must not drop the baseline messages table") + } +} + +// TestUp004MessagesNoReservedWords ensures no bare reserved words in new columns. +func TestUp004MessagesNoReservedWords(t *testing.T) { + up := read004UpMigration(t) + + // Check for bare "references" as column name + bareRefs := regexp.MustCompile(`(?im)^\s*REFERENCES\s+`) + if bareRefs.MatchString(up) { + lines := strings.Split(up, "\n") + for _, line := range lines { + trimmed := strings.TrimSpace(line) + upperLine := strings.ToUpper(trimmed) + if strings.HasPrefix(upperLine, "REFERENCES ") && !strings.Contains(upperLine, "FOREIGN KEY") { + t.Errorf("reserved word 'references' used as column name: %s", trimmed) + } + } + } +} diff --git a/internal/channels/dingtalk/handler.go b/internal/channels/dingtalk/handler.go index c0096d3..d068e3f 100644 --- a/internal/channels/dingtalk/handler.go +++ b/internal/channels/dingtalk/handler.go @@ -49,6 +49,10 @@ type QAAsker interface { Ask(ctx context.Context, req core.QARequest) (core.QAResult, error) } +type DeliveryRecorder interface { + RecordDelivery(ctx context.Context, update core.DeliveryUpdate) error +} + // StreamingQAAsker asks the QA backend with streaming observer support. type StreamingQAAsker interface { QAAsker @@ -257,6 +261,9 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req } return nil } + if result.Replayed && result.DeliveryStatus == core.DeliverySent { + return nil + } if h.card != nil { cardReq := CardRequest{ @@ -265,12 +272,18 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req ExternalChatID: req.ExternalChatID, Content: result.Answer, } - if _, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil { + if outTrackID, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil { + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "") return nil } log.Printf("DingTalk card delivery failed, falling back to markdown") } - return h.sendReply(ctx, webhookURL, result.Answer) + if err := h.sendReply(ctx, webhookURL, result.Answer); err != nil { + h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + return err + } + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + return nil } func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req core.QARequest, streamingCard StreamingCardSender, streamingQA StreamingQAAsker) error { @@ -291,7 +304,15 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co } return nil } - return h.sendReply(ctx, webhookURL, result.Answer) + if result.Replayed && result.DeliveryStatus == core.DeliverySent { + return nil + } + if replyErr := h.sendReply(ctx, webhookURL, result.Answer); replyErr != nil { + h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + return replyErr + } + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + return nil } // Streaming observer with the configured update throttle (500ms by default). @@ -314,9 +335,9 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co result, err := streamingQA.AskStream(ctx, req, observer) if err != nil || strings.TrimSpace(result.Answer) == "" { - return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, qaErrorReply, true) + return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, core.QAResult{}, qaErrorReply, true) } - return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result.Answer, false) + return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result, result.Answer, false) } func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, req core.QARequest, streamingCard StreamingCardSender, eventQA EventQAAsker) error { @@ -333,7 +354,15 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r if askErr != nil { return h.sendReply(ctx, webhookURL, qaErrorReply) } - return h.sendReply(ctx, webhookURL, result.Answer) + if result.Replayed && result.DeliveryStatus == core.DeliverySent { + return nil + } + if replyErr := h.sendReply(ctx, webhookURL, result.Answer); replyErr != nil { + h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + return replyErr + } + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + return nil } pump := newCardUpdatePump(ctx, h.cardUpdateInterval, func(updateCtx context.Context, content string) error { @@ -358,12 +387,12 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r pump.Stop() if err != nil || strings.TrimSpace(result.Answer) == "" { - return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, qaErrorReply, true) + return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, core.QAResult{}, qaErrorReply, true) } - return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result.Answer, false) + return h.finalizeCardOrMarkdown(ctx, webhookURL, streamingCard, outTrackID, result, result.Answer, false) } -func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string, streamingCard StreamingCardSender, outTrackID, content string, isError bool) error { +func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string, streamingCard StreamingCardSender, outTrackID string, result core.QAResult, content string, isError bool) error { timeout := h.finalDeliveryTimeout if timeout <= 0 { timeout = defaultHTTPTimeout @@ -375,6 +404,9 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string, updateErr := streamingCard.UpdateCard(finalCtx, outTrackID, content, true, isError) cancelFinal() if updateErr == nil { + if !isError { + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "") + } return nil } @@ -382,7 +414,45 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string, // a fresh delivery window rather than reusing the expired finalCtx. fallbackCtx, cancelFallback := context.WithTimeout(context.WithoutCancel(ctx), timeout) defer cancelFallback() - return h.sendReply(fallbackCtx, webhookURL, content) + if err := h.sendReply(fallbackCtx, webhookURL, content); err != nil { + if !isError { + h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + } + return err + } + if !isError { + h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + } + return nil +} + +func (h *Handler) recordDelivery(ctx context.Context, result core.QAResult, status, method, providerMessageID, failureCode string) { + if strings.TrimSpace(result.TurnID) == "" || strings.TrimSpace(result.AnswerMessageID) == "" { + return + } + recorder, ok := h.qa.(DeliveryRecorder) + if !ok { + return + } + timeout := h.finalDeliveryTimeout + if timeout <= 0 { + timeout = defaultHTTPTimeout + } + auditCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + defer cancel() + err := recorder.RecordDelivery(auditCtx, core.DeliveryUpdate{ + TenantKey: h.cfg.TenantKey, + Channel: h.cfg.Channel, + AnswerMessageID: result.AnswerMessageID, + CorrelationID: result.TurnID, + Status: status, + DeliveryMethod: method, + ProviderMessageID: providerMessageID, + FailureCode: failureCode, + }) + if err != nil { + log.Printf("DingTalk delivery audit update failed: status=%s method=%s", status, method) + } } func renderProgressEvent(event weknora.StreamEvent) string { diff --git a/internal/channels/dingtalk/handler_test.go b/internal/channels/dingtalk/handler_test.go index 11f118f..5814492 100644 --- a/internal/channels/dingtalk/handler_test.go +++ b/internal/channels/dingtalk/handler_test.go @@ -12,10 +12,12 @@ import ( ) type fakeQA struct { - called bool - lastReq core.QARequest - result core.QAResult - err error + called bool + lastReq core.QARequest + result core.QAResult + err error + deliveries []core.DeliveryUpdate + deliveryErr error } func (f *fakeQA) Ask(_ context.Context, req core.QARequest) (core.QAResult, error) { @@ -24,6 +26,11 @@ func (f *fakeQA) Ask(_ context.Context, req core.QARequest) (core.QAResult, erro return f.result, f.err } +func (f *fakeQA) RecordDelivery(_ context.Context, update core.DeliveryUpdate) error { + f.deliveries = append(f.deliveries, update) + return f.deliveryErr +} + type fakeReply struct { called bool lastURL string @@ -829,6 +836,8 @@ type fakeStreamingQA struct { streamCalled bool chunks []string // observer receives cumulated content after each chunk observerCaptured weknora.AnswerObserver + deliveries []core.DeliveryUpdate + deliveryErr error } func (f *fakeStreamingQA) Ask(_ context.Context, req core.QARequest) (core.QAResult, error) { @@ -851,6 +860,11 @@ func (f *fakeStreamingQA) AskStream(_ context.Context, req core.QARequest, obser return f.result, f.err } +func (f *fakeStreamingQA) RecordDelivery(_ context.Context, update core.DeliveryUpdate) error { + f.deliveries = append(f.deliveries, update) + return f.deliveryErr +} + // fakeStreamingCard implements both FinalCardSender and StreamingCardSender. type fakeStreamingCard struct { // FinalCardSender @@ -1660,3 +1674,125 @@ func TestHandleEventStreamingFinalTimeoutUsesFreshMarkdownContext(t *testing.T) t.Fatalf("Markdown fallback received expired context: %v", reply.contextErr) } } + +func deliveryQAResult() core.QAResult { + return core.QAResult{ + TurnID: "turn-delivery-1", AnswerMessageID: "answer-delivery-1", + DeliveryStatus: core.DeliveryGenerated, Answer: "delivery answer", + } +} + +func requireDelivery(t *testing.T, deliveries []core.DeliveryUpdate, status, method, providerID, failureCode string) { + t.Helper() + if len(deliveries) != 1 { + t.Fatalf("delivery updates=%d want=1: %+v", len(deliveries), deliveries) + } + got := deliveries[0] + if got.TenantKey != "tenant-a" || got.Channel != "dingtalk" || + got.CorrelationID != "turn-delivery-1" || got.AnswerMessageID != "answer-delivery-1" || + got.Status != status || got.DeliveryMethod != method || + got.ProviderMessageID != providerID || got.FailureCode != failureCode { + t.Fatalf("unexpected delivery update: %+v", got) + } +} + +func TestDeliveryAuditNonStreamingCardSuccess(t *testing.T) { + qa := &fakeQA{result: deliveryQAResult()} + h := newTestHandlerWithCard(t, qa, &fakeReply{}, &fakeCard{returnID: "track-delivery-1"}) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodCard, "track-delivery-1", "") +} + +func TestDeliveryAuditNonStreamingMarkdownFallbackAndFailure(t *testing.T) { + t.Run("markdown succeeds", func(t *testing.T) { + qa := &fakeQA{result: deliveryQAResult()} + h := newTestHandlerWithCard(t, qa, &fakeReply{}, &fakeCard{err: errors.New("card failed")}) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + }) + t.Run("both fail", func(t *testing.T) { + qa := &fakeQA{result: deliveryQAResult()} + h := newTestHandlerWithCard(t, qa, &fakeReply{err: errors.New("markdown failed")}, &fakeCard{err: errors.New("card failed")}) + if err := h.Handle(context.Background(), directTextCallback()); err == nil { + t.Fatal("expected delivery error") + } + requireDelivery(t, qa.deliveries, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + }) +} + +func TestDeliveryAuditFailureAfterSuccessfulSendDoesNotTriggerRedelivery(t *testing.T) { + qa := &fakeQA{result: deliveryQAResult(), deliveryErr: errors.New("database unavailable")} + reply := &fakeReply{} + card := &fakeCard{returnID: "track-delivery-1"} + h := newTestHandlerWithCard(t, qa, reply, card) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatalf("successful channel delivery must remain successful: %v", err) + } + if !card.called || reply.called || len(qa.deliveries) != 1 { + t.Fatalf("delivery was duplicated: card=%v markdown=%v audits=%d", card.called, reply.called, len(qa.deliveries)) + } +} + +func TestDeliveryAuditSuppressesSentReplayBeforeNonStreamingDelivery(t *testing.T) { + result := deliveryQAResult() + result.Replayed = true + result.DeliveryStatus = core.DeliverySent + qa := &fakeQA{result: result} + reply := &fakeReply{} + card := &fakeCard{returnID: "must-not-send"} + h := newTestHandlerWithCard(t, qa, reply, card) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + if card.called || reply.called || len(qa.deliveries) != 0 { + t.Fatalf("sent replay must have zero delivery side effects: card=%v markdown=%v audits=%d", card.called, reply.called, len(qa.deliveries)) + } +} + +func TestDeliveryAuditStreamingCardAndMarkdown(t *testing.T) { + t.Run("final card succeeds", func(t *testing.T) { + qa := &fakeStreamingQA{result: deliveryQAResult(), chunks: []string{"delivery answer"}} + card := &fakeStreamingCard{startID: "stream-track-1"} + h := newTestHandlerWithStreaming(t, qa, &fakeReply{}, card) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodCard, "stream-track-1", "") + }) + t.Run("final card fails markdown succeeds", func(t *testing.T) { + qa := &fakeStreamingQA{result: deliveryQAResult(), chunks: []string{"delivery answer"}} + card := &fakeStreamingCard{startID: "stream-track-2", updateErr: errors.New("update failed")} + h := newTestHandlerWithStreaming(t, qa, &fakeReply{}, card) + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodMarkdown, "", "") + }) + t.Run("final card and markdown fail", func(t *testing.T) { + qa := &fakeStreamingQA{result: deliveryQAResult(), chunks: []string{"delivery answer"}} + card := &fakeStreamingCard{startID: "stream-track-3", updateErr: errors.New("update failed")} + h := newTestHandlerWithStreaming(t, qa, &fakeReply{err: errors.New("markdown failed")}, card) + if err := h.Handle(context.Background(), directTextCallback()); err == nil { + t.Fatal("expected delivery error") + } + requireDelivery(t, qa.deliveries, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery) + }) +} + +func TestDeliveryAuditEventStreamingProductionPath(t *testing.T) { + qa := &fakeEventQA{ + fakeStreamingQA: fakeStreamingQA{result: deliveryQAResult()}, + events: []weknora.StreamEvent{{ResponseType: "answer", FullAnswer: "delivery answer", Done: true}}, + } + card := &fakeStreamingCard{startID: "event-track-1"} + h := newTestHandlerWithStreaming(t, qa, &fakeReply{}, card) + h.cardUpdateInterval = time.Hour + if err := h.Handle(context.Background(), directTextCallback()); err != nil { + t.Fatal(err) + } + requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodCard, "event-track-1", "") +} diff --git a/internal/core/delivery_test.go b/internal/core/delivery_test.go new file mode 100644 index 0000000..fbf8268 --- /dev/null +++ b/internal/core/delivery_test.go @@ -0,0 +1,140 @@ +package core + +import ( + "context" + "encoding/json" + "strings" + "testing" +) + +func TestDeliveryStatusConstants(t *testing.T) { + // Verify delivery status constants exist + if DeliveryGenerated != "generated" { + t.Errorf("DeliveryGenerated = %q, want 'generated'", DeliveryGenerated) + } + if DeliverySent != "sent" { + t.Errorf("DeliverySent = %q, want 'sent'", DeliverySent) + } + if DeliveryFailed != "failed" { + t.Errorf("DeliveryFailed = %q, want 'failed'", DeliveryFailed) + } +} + +func TestDeliveryMethodConstants(t *testing.T) { + if DeliveryMethodCard != "card" { + t.Errorf("DeliveryMethodCard = %q, want 'card'", DeliveryMethodCard) + } + if DeliveryMethodMarkdown != "markdown" { + t.Errorf("DeliveryMethodMarkdown = %q, want 'markdown'", DeliveryMethodMarkdown) + } +} + +func TestDeliveryFailureCodeConstants(t *testing.T) { + if DeliveryFailureChannelDelivery != "channel_delivery_failed" { + t.Errorf("DeliveryFailureChannelDelivery = %q, want 'channel_delivery_failed'", + DeliveryFailureChannelDelivery) + } +} + +func TestDeliveryUpdateFields(t *testing.T) { + du := DeliveryUpdate{ + TenantKey: "t1", + Channel: "dingtalk", + AnswerMessageID: "msg-1", + CorrelationID: "turn-1", + Status: DeliverySent, + DeliveryMethod: DeliveryMethodCard, + ProviderMessageID: "out-track-123", + FailureCode: "", + } + + if du.TenantKey != "t1" { + t.Error("DeliveryUpdate must have TenantKey") + } + if du.Channel != "dingtalk" { + t.Error("DeliveryUpdate must have Channel") + } + if du.AnswerMessageID != "msg-1" { + t.Error("DeliveryUpdate must have AnswerMessageID") + } + if du.CorrelationID != "turn-1" { + t.Error("DeliveryUpdate must have CorrelationID") + } + if du.Status != DeliverySent { + t.Error("DeliveryUpdate must have Status") + } + if du.DeliveryMethod != DeliveryMethodCard { + t.Error("DeliveryUpdate must have DeliveryMethod") + } + if du.ProviderMessageID != "out-track-123" { + t.Error("DeliveryUpdate must have ProviderMessageID") + } + if du.FailureCode != "" { + t.Error("DeliveryUpdate FailureCode should be empty when not failed") + } +} + +func TestDeliveryUpdateNoSecrets(t *testing.T) { + raw, err := json.Marshal(DeliveryUpdate{}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + s := string(raw) + for _, forbidden := range []string{"sessionWebhook", "token", "webhook", "raw_response", "raw_error", "outTrackID"} { + if containsCaseInsensitive(s, forbidden) { + t.Errorf("DeliveryUpdate JSON must not contain %q: %s", forbidden, s) + } + } +} + +func TestQAResultHasAnswerMessageID(t *testing.T) { + qr := QAResult{ + AnswerMessageID: "answer-msg-1", + DeliveryStatus: DeliveryGenerated, + } + if qr.AnswerMessageID != "answer-msg-1" { + t.Error("QAResult must have AnswerMessageID") + } + if qr.DeliveryStatus != DeliveryGenerated { + t.Error("QAResult must have DeliveryStatus") + } +} + +func TestTurnHasDeliveryFields(t *testing.T) { + turn := Turn{ + AnswerMessageID: "ans-1", + } + if turn.AnswerMessageID != "ans-1" { + t.Error("Turn must have AnswerMessageID") + } + // DeliveryStatus and DeliveryMethod fields on Turn for loadTurn recovery + // These are set by the store when loading + turn.AnswerDeliveryStatus = DeliverySent + turn.AnswerDeliveryMethod = DeliveryMethodCard + if turn.AnswerDeliveryStatus != DeliverySent { + t.Error("Turn must have AnswerDeliveryStatus") + } + if turn.AnswerDeliveryMethod != DeliveryMethodCard { + t.Error("Turn must have AnswerDeliveryMethod") + } +} + +func TestRepositoryHasRecordDelivery(t *testing.T) { + var repository Repository + var _ interface { + RecordDelivery(context.Context, DeliveryUpdate) error + } = repository +} + +func TestErrDeliveryUpdateFailedExists(t *testing.T) { + if ErrDeliveryUpdateFailed == nil { + t.Error("ErrDeliveryUpdateFailed must be defined") + } + if ErrDeliveryUpdateFailed.Error() == "" { + t.Error("ErrDeliveryUpdateFailed must have a message") + } +} + +func containsCaseInsensitive(s, substr string) bool { + return strings.Contains(strings.ToLower(s), strings.ToLower(substr)) +} diff --git a/internal/core/types.go b/internal/core/types.go index 736f634..979a691 100644 --- a/internal/core/types.go +++ b/internal/core/types.go @@ -60,6 +60,8 @@ type Reference struct { type QAResult struct { TurnID string `json:"turn_id"` + AnswerMessageID string `json:"answer_message_id,omitempty"` + DeliveryStatus string `json:"delivery_status,omitempty"` WeKnoraSessionID string `json:"weknora_session_id"` WeKnoraRequestID string `json:"weknora_request_id,omitempty"` Answer string `json:"answer"` @@ -78,33 +80,69 @@ type SessionMapping struct { } type Turn struct { - ID string + ID string + TenantKey string + Channel string + ExternalMessageID string + SessionKey string + WeKnoraSessionID string + WeKnoraRequestID string + AgentID string + ConversationType ConversationType + ExternalChatID string + ExternalUserID string + SenderName string + MessageType string + ActorID string + ConversationID string + RequestMessageID string + AnswerMessageID string + AnswerDeliveryStatus string + AnswerDeliveryMethod string + Question string + Answer string + References []Reference + Media []MediaAudit + Status string + ErrorMessage string +} + +// Delivery status constants for outbound messages. +const ( + DeliveryGenerated = "generated" + DeliverySent = "sent" + DeliveryFailed = "failed" +) + +// Delivery method constants. +const ( + DeliveryMethodCard = "card" + DeliveryMethodMarkdown = "markdown" +) + +// DeliveryFailureCode constants. +const ( + DeliveryFailureChannelDelivery = "channel_delivery_failed" +) + +// DeliveryUpdate carries the minimum safe fields needed to record a delivery outcome. +// It must never contain sessionWebhook, tokens, raw API responses, or raw error text. +type DeliveryUpdate struct { TenantKey string Channel string - ExternalMessageID string - SessionKey string - WeKnoraSessionID string - WeKnoraRequestID string - AgentID string - ConversationType ConversationType - ExternalChatID string - ExternalUserID string - SenderName string - MessageType string - ActorID string - ConversationID string - RequestMessageID string AnswerMessageID string - Question string - Answer string - References []Reference - Media []MediaAudit + CorrelationID string Status string - ErrorMessage string + DeliveryMethod string + ProviderMessageID string + FailureCode string } +var ErrDeliveryUpdateFailed = errors.New("delivery update failed: RowsAffected != 1") + type Repository interface { BeginTurn(ctx context.Context, turn Turn) (Turn, bool, error) ResolveSession(ctx context.Context, mapping SessionMapping, create func(context.Context) (string, error)) (string, error) FinishTurn(ctx context.Context, turn Turn) error + RecordDelivery(ctx context.Context, du DeliveryUpdate) error } diff --git a/internal/qa/service.go b/internal/qa/service.go index bcf0139..9b31697 100644 --- a/internal/qa/service.go +++ b/internal/qa/service.go @@ -59,6 +59,10 @@ func (s *Service) AskEvents(ctx context.Context, request core.QARequest, observe return s.ask(ctx, request, nil, observer) } +func (s *Service) RecordDelivery(ctx context.Context, update core.DeliveryUpdate) error { + return s.repository.RecordDelivery(ctx, update) +} + func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserver weknora.AnswerObserver, eventObserver weknora.StreamObserver) (core.QAResult, error) { request = normalizeRequestIdentity(request) if err := validateRequest(request); err != nil { @@ -115,6 +119,7 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve if stored.Status == "succeeded" { return core.QAResult{TurnID: stored.ID, WeKnoraSessionID: stored.WeKnoraSessionID, WeKnoraRequestID: stored.WeKnoraRequestID, Answer: stored.Answer, + AnswerMessageID: stored.AnswerMessageID, DeliveryStatus: stored.AnswerDeliveryStatus, References: stored.References, Replayed: true}, nil } return core.QAResult{}, core.ErrTurnInProgress @@ -202,7 +207,8 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve if err := s.repository.FinishTurn(ctx, turn); err != nil { return core.QAResult{}, fmt.Errorf("finish turn: %w", err) } - return core.QAResult{TurnID: turn.ID, WeKnoraSessionID: sessionID, + return core.QAResult{TurnID: turn.ID, AnswerMessageID: turn.AnswerMessageID, + DeliveryStatus: core.DeliveryGenerated, WeKnoraSessionID: sessionID, WeKnoraRequestID: chat.RequestID, Answer: chat.Answer, References: chat.References}, nil } diff --git a/internal/qa/service_test.go b/internal/qa/service_test.go index 038f992..346de1e 100644 --- a/internal/qa/service_test.go +++ b/internal/qa/service_test.go @@ -18,6 +18,7 @@ type memoryRepository struct { sessions map[string]string beginTurns []core.Turn finishTurns []core.Turn + deliveries []core.DeliveryUpdate } func newMemoryRepository() *memoryRepository { @@ -60,6 +61,12 @@ func (r *memoryRepository) FinishTurn(_ context.Context, turn core.Turn) error { r.turns[turn.TenantKey+":"+turn.Channel+":"+turn.ExternalMessageID] = turn return nil } +func (r *memoryRepository) RecordDelivery(_ context.Context, update core.DeliveryUpdate) error { + r.mu.Lock() + defer r.mu.Unlock() + r.deliveries = append(r.deliveries, update) + return nil +} type fixedResolver struct{} @@ -583,3 +590,56 @@ func TestAskRejectsNonQAMessageTypes(t *testing.T) { }) } } + +func TestAskReturnsGeneratedDeliveryIdentityAndReplay(t *testing.T) { + repo := newMemoryRepository() + service, err := New(repo, fixedResolver{}, &fakeWeKnora{}, session.GroupShared) + if err != nil { + t.Fatal(err) + } + req := core.QARequest{ + TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect, + ExternalUserID: "user-1", ExternalMessageID: "message-delivery", Query: "question", + } + first, err := service.Ask(context.Background(), req) + if err != nil { + t.Fatal(err) + } + if first.AnswerMessageID == "" || first.DeliveryStatus != core.DeliveryGenerated { + t.Fatalf("new result delivery identity incomplete: %+v", first) + } + + repo.mu.Lock() + key := req.TenantKey + ":" + req.Channel + ":" + req.ExternalMessageID + stored := repo.turns[key] + stored.AnswerDeliveryStatus = core.DeliverySent + repo.turns[key] = stored + repo.mu.Unlock() + replay, err := service.Ask(context.Background(), req) + if err != nil { + t.Fatal(err) + } + if !replay.Replayed || replay.AnswerMessageID != first.AnswerMessageID || replay.DeliveryStatus != core.DeliverySent { + t.Fatalf("replay delivery identity incomplete: first=%+v replay=%+v", first, replay) + } +} + +func TestRecordDeliveryDelegatesToRepository(t *testing.T) { + repo := newMemoryRepository() + service, err := New(repo, fixedResolver{}, &fakeWeKnora{}, session.GroupShared) + if err != nil { + t.Fatal(err) + } + update := core.DeliveryUpdate{ + TenantKey: "tenant", Channel: "dingtalk", AnswerMessageID: "answer-1", CorrelationID: "turn-1", + Status: core.DeliverySent, DeliveryMethod: core.DeliveryMethodCard, ProviderMessageID: "track-1", + } + if err := service.RecordDelivery(context.Background(), update); err != nil { + t.Fatal(err) + } + repo.mu.Lock() + defer repo.mu.Unlock() + if len(repo.deliveries) != 1 || repo.deliveries[0] != update { + t.Fatalf("delivery update not delegated: %+v", repo.deliveries) + } +} diff --git a/internal/store/postgres/delivery_audit_test.go b/internal/store/postgres/delivery_audit_test.go new file mode 100644 index 0000000..9b2b9fe --- /dev/null +++ b/internal/store/postgres/delivery_audit_test.go @@ -0,0 +1,91 @@ +package postgres + +import ( + "os" + "strings" + "testing" + + "gitea.zheromiku.top/Weknora/dobotai-platform/internal/core" +) + +func TestValidateDeliveryUpdate(t *testing.T) { + validSent := core.DeliveryUpdate{ + TenantKey: "tenant", Channel: "dingtalk", AnswerMessageID: "answer-1", CorrelationID: "turn-1", + Status: core.DeliverySent, DeliveryMethod: core.DeliveryMethodCard, ProviderMessageID: "track-1", + } + validFailed := core.DeliveryUpdate{ + TenantKey: "tenant", Channel: "dingtalk", AnswerMessageID: "answer-1", CorrelationID: "turn-1", + Status: core.DeliveryFailed, FailureCode: core.DeliveryFailureChannelDelivery, + } + tests := []struct { + name string + update core.DeliveryUpdate + wantErr bool + }{ + {name: "sent-card", update: validSent}, + {name: "sent-markdown", update: func() core.DeliveryUpdate { + u := validSent + u.DeliveryMethod = core.DeliveryMethodMarkdown + u.ProviderMessageID = "" + return u + }()}, + {name: "failed", update: validFailed}, + {name: "missing-tenant", update: func() core.DeliveryUpdate { u := validSent; u.TenantKey = ""; return u }(), wantErr: true}, + {name: "missing-correlation", update: func() core.DeliveryUpdate { u := validSent; u.CorrelationID = ""; return u }(), wantErr: true}, + {name: "generated-not-terminal", update: func() core.DeliveryUpdate { u := validSent; u.Status = core.DeliveryGenerated; return u }(), wantErr: true}, + {name: "sent-needs-method", update: func() core.DeliveryUpdate { u := validSent; u.DeliveryMethod = ""; return u }(), wantErr: true}, + {name: "sent-no-failure", update: func() core.DeliveryUpdate { + u := validSent + u.FailureCode = core.DeliveryFailureChannelDelivery + return u + }(), wantErr: true}, + {name: "failed-needs-safe-code", update: func() core.DeliveryUpdate { u := validFailed; u.FailureCode = "raw timeout from webhook"; return u }(), wantErr: true}, + {name: "failed-no-provider-id", update: func() core.DeliveryUpdate { u := validFailed; u.ProviderMessageID = "track"; return u }(), wantErr: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := validateDeliveryUpdate(tc.update) + if (err != nil) != tc.wantErr { + t.Fatalf("validateDeliveryUpdate error=%v wantErr=%v", err, tc.wantErr) + } + }) + } +} + +func TestRecordDeliverySQLIsTenantSafeAndMonotonic(t *testing.T) { + data, err := os.ReadFile("store.go") + if err != nil { + t.Fatal(err) + } + src := strings.ToUpper(string(data)) + for _, fragment := range []string{ + "UPDATE MESSAGES AS M", "FROM TURNS T", "M.TENANT_KEY=$1", "M.CHANNEL=$2", + "M.ID=$3", "M.CORRELATION_ID=$4", "T.ANSWER_MESSAGE_ID=M.ID", + "M.DIRECTION='OUTBOUND'", "M.ROLE='ASSISTANT'", "DELIVERED_AT=CASE", + "COALESCE(M.DELIVERED_AT,NOW())", "M.STATUS IN ('GENERATED','FAILED','SENT')", + "$5='FAILED' AND (M.STATUS='GENERATED' OR M.STATUS='FAILED')", "ROWSAFFECTED", + } { + if !strings.Contains(src, fragment) { + t.Errorf("RecordDelivery SQL missing %q", fragment) + } + } + if strings.Contains(src, "DELIVEREDAT") || strings.Contains(src, "DELIVERED_AT=$") { + t.Error("delivery timestamp must be generated by PostgreSQL NOW(), not supplied by caller") + } +} + +func TestLoadTurnRestoresAnswerDeliveryState(t *testing.T) { + data, err := os.ReadFile("store.go") + if err != nil { + t.Fatal(err) + } + src := strings.ToUpper(string(data)) + for _, fragment := range []string{ + "LEFT JOIN MESSAGES AM ON AM.ID=T.ANSWER_MESSAGE_ID", + "AM.STATUS", "AM.DELIVERY_METHOD", "ANSWERDELIVERYSTATUS", "ANSWERDELIVERYMETHOD", + } { + if !strings.Contains(src, fragment) { + t.Errorf("loadTurn delivery restore missing %q", fragment) + } + } +} diff --git a/internal/store/postgres/store.go b/internal/store/postgres/store.go index 614994f..b98ffab 100644 --- a/internal/store/postgres/store.go +++ b/internal/store/postgres/store.go @@ -186,6 +186,40 @@ func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error { return tx.Commit(ctx) } +func (s *Store) RecordDelivery(ctx context.Context, update core.DeliveryUpdate) error { + if err := validateDeliveryUpdate(update); err != nil { + return err + } + tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + + command, err := tx.Exec(ctx, `UPDATE messages AS m SET + status=$5, + delivery_method=CASE WHEN $5='sent' THEN $6 ELSE '' END, + provider_message_id=CASE WHEN $5='sent' THEN $7 ELSE '' END, + failure_code=CASE WHEN $5='failed' THEN $8 ELSE '' END, + delivered_at=CASE WHEN $5='sent' THEN COALESCE(m.delivered_at,NOW()) ELSE NULL END, + updated_at=NOW() + FROM turns t + WHERE m.tenant_key=$1 AND m.channel=$2 AND m.id=$3 AND m.correlation_id=$4 + AND m.direction='outbound' AND m.role='assistant' + AND t.id=$4 AND t.tenant_key=$1 AND t.channel=$2 AND t.answer_message_id=m.id + AND (($5='sent' AND m.status IN ('generated','failed','sent')) + OR ($5='failed' AND (m.status='generated' OR m.status='failed')))`, + update.TenantKey, update.Channel, update.AnswerMessageID, update.CorrelationID, + update.Status, update.DeliveryMethod, update.ProviderMessageID, update.FailureCode) + if err != nil { + return err + } + if command.RowsAffected() != 1 { + return core.ErrDeliveryUpdateFailed + } + return tx.Commit(ctx) +} + type rowQuerier interface { QueryRow(context.Context, string, ...any) pgx.Row } @@ -201,14 +235,17 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes var agentID, requestMessageID, answerMessageID *string var actorID, conversationID, messageType, actorName *string var conversationType, externalConversationID, externalActorID *string + var answerDeliveryStatus, answerDeliveryMethod *string err := query.QueryRow(ctx, `SELECT t.id,t.tenant_key,t.channel,t.external_message_id,t.session_key,t.weknora_session_id, t.weknora_request_id,t.question,t.answer,t.references_json,t.media_json,t.status,t.error_message, t.agent_id,t.request_message_id,t.answer_message_id, m.actor_id,m.conversation_id,m.message_type,m.actor_name_snapshot, - c.conversation_type,c.external_conversation_id,a.external_actor_id + c.conversation_type,c.external_conversation_id,a.external_actor_id, + am.status,am.delivery_method FROM turns t LEFT JOIN messages m ON m.id=t.request_message_id + LEFT JOIN messages am ON am.id=t.answer_message_id LEFT JOIN channel_conversations c ON c.id=m.conversation_id LEFT JOIN channel_actors a ON a.id=m.actor_id WHERE t.tenant_key=$1 AND t.channel=$2 AND t.external_message_id=$3`, tenantKey, channel, messageID).Scan( @@ -216,7 +253,7 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes &turn.WeKnoraSessionID, &turn.WeKnoraRequestID, &turn.Question, &turn.Answer, &references, &mediaJSON, &turn.Status, &turn.ErrorMessage, &agentID, &requestMessageID, &answerMessageID, &actorID, &conversationID, &messageType, &actorName, &conversationType, - &externalConversationID, &externalActorID) + &externalConversationID, &externalActorID, &answerDeliveryStatus, &answerDeliveryMethod) if err != nil { return core.Turn{}, err } @@ -229,6 +266,12 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes if answerMessageID != nil { turn.AnswerMessageID = *answerMessageID } + if answerDeliveryStatus != nil { + turn.AnswerDeliveryStatus = *answerDeliveryStatus + } + if answerDeliveryMethod != nil { + turn.AnswerDeliveryMethod = *answerDeliveryMethod + } if actorID != nil { turn.ActorID = *actorID } @@ -269,6 +312,33 @@ func shouldCreateAnswerMessage(turn core.Turn) bool { strings.TrimSpace(turn.AnswerMessageID) != "" } +func validateDeliveryUpdate(update core.DeliveryUpdate) error { + if strings.TrimSpace(update.TenantKey) == "" || strings.TrimSpace(update.Channel) == "" || + strings.TrimSpace(update.AnswerMessageID) == "" || strings.TrimSpace(update.CorrelationID) == "" { + return errors.New("delivery update identity is required") + } + switch update.Status { + case core.DeliverySent: + if update.DeliveryMethod != core.DeliveryMethodCard && update.DeliveryMethod != core.DeliveryMethodMarkdown { + return errors.New("sent delivery requires a supported method") + } + if update.FailureCode != "" { + return errors.New("sent delivery cannot have a failure code") + } + if update.DeliveryMethod == core.DeliveryMethodMarkdown && update.ProviderMessageID != "" { + return errors.New("markdown delivery cannot have a provider message id") + } + case core.DeliveryFailed: + if update.DeliveryMethod != "" || update.ProviderMessageID != "" || + update.FailureCode != core.DeliveryFailureChannelDelivery { + return errors.New("failed delivery requires a safe failure code only") + } + default: + return errors.New("delivery status must be sent or failed") + } + return nil +} + func externalConversationID(turn core.Turn) string { if externalChatID := strings.TrimSpace(turn.ExternalChatID); externalChatID != "" { return externalChatID