Version: 0.5.8.dev.260315

♻️ refactor(agent): 拆分 agentsvc，并增强 quicknote/outbox 注释与可维护性 - 📦 将 Agent 服务实现从 `service` 根目录迁移到 `service/agentsvc`，包含 `agent.go`、`agent_quick_note.go` 及相关测试 - 🔌 新增 service 层兼容桥接 `agent_bridge.go`，保持 `service.NewAgentService` 与 `*service.AgentService` 现有调用方式不变 - 📝 为 `quicknote` 补充高密度中文步骤化注释，覆盖 `graph` / `runner` / `nodes` / `tool` / `state` / `prompt`，明确职责边界、分支条件、重试与兜底策略 - 🧭 为 `infra/outbox` 与 service agent 链路补充详细中文注释，覆盖状态机流转、幂等处理、失败回写与异步持久化语义 - ✅ 统一格式化相关文件，并通过全量后端测试：`go test ./...` 📝 chore(docs): 更新 AGENTS.md 注释强制规范 - 📚 追加“注释规范（强制）”与“注释风格示例” - ✍️ 明确复杂逻辑必须使用步骤化注释、跨文件调用需写调用目的、注释需同步维护
2026-03-15 18:08:33 +08:00
parent c689af56c8
commit 7603a7561a
22 changed files with 1009 additions and 429 deletions
--- a/backend/infra/kafka/admin.go
+++ b/backend/infra/kafka/admin.go
@@ -0,0 +1,78 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	segmentkafka "github.com/segmentio/kafka-go"
+)
+
+// WaitTopicReady 在指定超时时间内等待 Kafka topic 可用。
+// 背景：初次部署时 broker 可能已启动，但 topic/partition 还没就绪。
+// 这里启动前先探测，可减少“应用已启动但实际无法消费”的静默窗口。
+func WaitTopicReady(parent context.Context, brokers []string, topic string, timeout time.Duration) error {
+	if len(brokers) == 0 {
+		return errors.New("kafka brokers is empty")
+	}
+	if topic == "" {
+		return errors.New("kafka topic is empty")
+	}
+	if timeout <= 0 {
+		timeout = 30 * time.Second
+	}
+
+	ctx, cancel := context.WithTimeout(parent, timeout)
+	defer cancel()
+
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	var lastErr error
+	for {
+		if err := probeTopic(ctx, brokers, topic); err == nil {
+			return nil
+		} else {
+			lastErr = err
+		}
+
+		select {
+		case <-ctx.Done():
+			if lastErr != nil {
+				return fmt.Errorf("wait topic ready timeout, topic=%s: %w", topic, lastErr)
+			}
+			return fmt.Errorf("wait topic ready timeout, topic=%s", topic)
+		case <-ticker.C:
+		}
+	}
+}
+
+// probeTopic 轮询所有 broker，只要任一 broker 能读到 topic 分区信息即视为就绪。
+func probeTopic(ctx context.Context, brokers []string, topic string) error {
+	var lastErr error
+	for _, broker := range brokers {
+		conn, err := segmentkafka.DialContext(ctx, "tcp", broker)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+
+		partitions, readErr := conn.ReadPartitions(topic)
+		_ = conn.Close()
+		if readErr != nil {
+			lastErr = readErr
+			continue
+		}
+		if len(partitions) == 0 {
+			lastErr = fmt.Errorf("topic %s has no partitions yet", topic)
+			continue
+		}
+		return nil
+	}
+
+	if lastErr != nil {
+		return lastErr
+	}
+	return errors.New("unable to probe topic readiness")
+}
--- a/backend/infra/kafka/config.go
+++ b/backend/infra/kafka/config.go
@@ -0,0 +1,63 @@
+package kafka
+
+import (
+	"strings"
+	"time"
+
+	"github.com/spf13/viper"
+)
+
+const (
+	DefaultTopic = "smartflow.agent.outbox"
+	DefaultGroup = "smartflow-agent-outbox-consumer"
+)
+
+// Config 描述 outbox 异步链路所需的 Kafka 配置。
+// 说明：这些参数同时影响“发送端（producer）”与“消费端（consumer）”。
+type Config struct {
+	Enabled bool
+	Brokers []string
+	Topic   string
+	GroupID string
+	// RetryScanInterval/RetryBatchSize/MaxRetry 作用于 outbox 扫描与失败重试。
+	RetryScanInterval time.Duration
+	RetryBatchSize    int
+	MaxRetry          int
+}
+
+// LoadConfig 从配置中心读取 Kafka 配置，并做兜底默认值。
+// 兼容性：优先读取 kafka.brokers（数组），为空时降级读取 kafka.broker（单值）。
+func LoadConfig() Config {
+	brokers := viper.GetStringSlice("kafka.brokers")
+	if len(brokers) == 0 {
+		single := strings.TrimSpace(viper.GetString("kafka.broker"))
+		if single != "" {
+			brokers = []string{single}
+		}
+	}
+	cfg := Config{
+		Enabled:           viper.GetBool("kafka.enabled"),
+		Brokers:           brokers,
+		Topic:             strings.TrimSpace(viper.GetString("kafka.topic")),
+		GroupID:           strings.TrimSpace(viper.GetString("kafka.groupID")),
+		RetryScanInterval: viper.GetDuration("kafka.retryScanInterval"),
+		RetryBatchSize:    viper.GetInt("kafka.retryBatchSize"),
+		MaxRetry:          viper.GetInt("kafka.maxRetry"),
+	}
+	if cfg.Topic == "" {
+		cfg.Topic = DefaultTopic
+	}
+	if cfg.GroupID == "" {
+		cfg.GroupID = DefaultGroup
+	}
+	if cfg.RetryScanInterval <= 0 {
+		cfg.RetryScanInterval = time.Second
+	}
+	if cfg.RetryBatchSize <= 0 {
+		cfg.RetryBatchSize = 100
+	}
+	if cfg.MaxRetry <= 0 {
+		cfg.MaxRetry = 20
+	}
+	return cfg
+}
--- a/backend/infra/kafka/consumer.go
+++ b/backend/infra/kafka/consumer.go
@@ -0,0 +1,50 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+
+	segmentkafka "github.com/segmentio/kafka-go"
+)
+
+type Consumer struct {
+	reader *segmentkafka.Reader
+}
+
+func NewConsumer(cfg Config) (*Consumer, error) {
+	if len(cfg.Brokers) == 0 {
+		return nil, errors.New("kafka brokers 未配置")
+	}
+	reader := segmentkafka.NewReader(segmentkafka.ReaderConfig{
+		Brokers:        cfg.Brokers,
+		Topic:          cfg.Topic,
+		GroupID:        cfg.GroupID,
+		MinBytes:       1,
+		MaxBytes:       10e6,
+		CommitInterval: 0,
+		StartOffset:    segmentkafka.FirstOffset,
+	})
+	return &Consumer{reader: reader}, nil
+}
+
+// Dequeue 从 Kafka 拉取一条消息（不自动提交 offset）。
+func (c *Consumer) Dequeue(ctx context.Context) (segmentkafka.Message, error) {
+	if c == nil || c.reader == nil {
+		return segmentkafka.Message{}, errors.New("kafka consumer 未初始化")
+	}
+	return c.reader.FetchMessage(ctx)
+}
+
+func (c *Consumer) Commit(ctx context.Context, msg segmentkafka.Message) error {
+	if c == nil || c.reader == nil {
+		return errors.New("kafka consumer 未初始化")
+	}
+	return c.reader.CommitMessages(ctx, msg)
+}
+
+func (c *Consumer) Close() error {
+	if c == nil || c.reader == nil {
+		return nil
+	}
+	return c.reader.Close()
+}
--- a/backend/infra/kafka/envelope.go
+++ b/backend/infra/kafka/envelope.go
@@ -0,0 +1,14 @@
+package kafka
+
+import "encoding/json"
+
+// Envelope 是 outbox 投递到 Kafka 的统一包裹结构。
+// 设计目的：
+// 1) 消费端先拿到 outbox_id，可直接回写状态；
+// 2) biz_type 做分发，支持后续扩展更多异步业务；
+// 3) payload 保持原始 JSON，按业务类型再反序列化。
+type Envelope struct {
+	OutboxID int64           `json:"outbox_id"`
+	BizType  string          `json:"biz_type"`
+	Payload  json.RawMessage `json:"payload"`
+}
--- a/backend/infra/kafka/producer.go
+++ b/backend/infra/kafka/producer.go
@@ -0,0 +1,45 @@
+package kafka
+
+import (
+	"context"
+	"errors"
+
+	segmentkafka "github.com/segmentio/kafka-go"
+)
+
+type Producer struct {
+	writer *segmentkafka.Writer
+}
+
+func NewProducer(cfg Config) (*Producer, error) {
+	if len(cfg.Brokers) == 0 {
+		return nil, errors.New("kafka brokers 未配置")
+	}
+	writer := &segmentkafka.Writer{
+		Addr:         segmentkafka.TCP(cfg.Brokers...),
+		Balancer:     &segmentkafka.Hash{},
+		RequiredAcks: segmentkafka.RequireOne,
+		Async:        false,
+	}
+	return &Producer{writer: writer}, nil
+}
+
+// Enqueue 将消息写入 Kafka。
+func (p *Producer) Enqueue(ctx context.Context, topic, key string, value []byte) error {
+	if p == nil || p.writer == nil {
+		return errors.New("kafka producer 未初始化")
+	}
+	msg := segmentkafka.Message{
+		Topic: topic,
+		Key:   []byte(key),
+		Value: value,
+	}
+	return p.writer.WriteMessages(ctx, msg)
+}
+
+func (p *Producer) Close() error {
+	if p == nil || p.writer == nil {
+		return nil
+	}
+	return p.writer.Close()
+}
--- a/backend/infra/outbox/chat_history_async.go
+++ b/backend/infra/outbox/chat_history_async.go
@@ -0,0 +1,91 @@
+package outbox
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
+	"github.com/LoveLosita/smartflow/backend/model"
+)
+
+// ChatHistoryAsync 是“聊天记录异步持久化”的业务适配器。
+//
+// 设计目的：
+// 1) 让业务层只调用 EnqueueChatHistoryPersist，而不感知扫描/投递/消费细节；
+// 2) 保持现有 Agent 代码调用习惯，降低改造面；
+// 3) 把具体的 outbox+kafka 主流程彻底收敛到 infra。
+type ChatHistoryAsync struct {
+	engine *Engine
+}
+
+// NewChatHistoryAsync 创建聊天记录异步适配器并注册处理器。
+//
+// 处理器职责：
+// 1) 从 envelope payload 解析聊天载荷；
+// 2) 调用仓储“落库并标记 consumed”；
+// 3) 解析失败时标记 dead（不可恢复错误），避免无意义重试。
+func NewChatHistoryAsync(repo *Repository, cfg kafkabus.Config) (*ChatHistoryAsync, error) {
+	// 1. 先创建通用引擎，内部会按 cfg.Enabled 决定是否启用。
+	engine, err := NewEngine(repo, cfg)
+	if err != nil {
+		return nil, err
+	}
+	if engine == nil {
+		// 2. 异步开关关闭：返回 nil 交给上层走同步降级路径。
+		return nil, nil
+	}
+
+	// 3. 注册“聊天记录持久化”业务处理器。
+	//    该处理器只做三件事：
+	//    3.1 解析 payload；
+	//    3.2 调仓储落库并推进 consumed；
+	//    3.3 遇到不可恢复错误时标记 dead。
+	if err = engine.RegisterHandler(model.OutboxBizTypeChatHistoryPersist, func(ctx context.Context, envelope kafkabus.Envelope) error {
+		var payload model.ChatHistoryPersistPayload
+		if unmarshalErr := json.Unmarshal(envelope.Payload, &payload); unmarshalErr != nil {
+			_ = repo.MarkDead(ctx, envelope.OutboxID, "解析聊天持久化载荷失败: "+unmarshalErr.Error())
+			// 返回 nil：该错误已被标记为 dead，不需要再走重试。
+			return nil
+		}
+		// 返回 error：由 engine 统一标记为 retry 并提交 offset。
+		return repo.PersistChatHistoryAndMarkConsumed(ctx, envelope.OutboxID, payload)
+	}); err != nil {
+		// 4. 注册失败时回收已创建的引擎资源，防止泄漏。
+		engine.Close()
+		return nil, err
+	}
+
+	// 5. 返回业务适配器，对业务层暴露“更语义化”的调用入口。
+	return &ChatHistoryAsync{engine: engine}, nil
+}
+
+// Start 启动异步引擎（扫描 + 消费）。
+func (a *ChatHistoryAsync) Start(ctx context.Context) {
+	// 允许在未初始化（例如异步关闭）时被安全调用。
+	if a == nil || a.engine == nil {
+		return
+	}
+	a.engine.Start(ctx)
+}
+
+// Close 关闭异步引擎资源。
+func (a *ChatHistoryAsync) Close() {
+	// 允许在未初始化（例如异步关闭）时被安全调用。
+	if a == nil || a.engine == nil {
+		return
+	}
+	a.engine.Close()
+}
+
+// EnqueueChatHistoryPersist 将聊天记录持久化请求写入 outbox。
+// 该方法是业务层唯一需要调用的入口。
+func (a *ChatHistoryAsync) EnqueueChatHistoryPersist(ctx context.Context, payload model.ChatHistoryPersistPayload) error {
+	// 1. 若引擎未初始化，说明启动配置有问题或异步功能未启用。
+	//    这里显式返回错误，交由业务层按需降级/告警。
+	if a == nil || a.engine == nil {
+		return errors.New("chat history async is not initialized")
+	}
+	// 2. 以 conversation_id 作为 messageKey，尽量让同会话消息落在稳定分区。
+	return a.engine.Enqueue(ctx, model.OutboxBizTypeChatHistoryPersist, payload.ConversationID, payload)
+}
--- a/backend/infra/outbox/engine.go
+++ b/backend/infra/outbox/engine.go
@@ -0,0 +1,333 @@
+package outbox
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
+	"github.com/LoveLosita/smartflow/backend/model"
+	segmentkafka "github.com/segmentio/kafka-go"
+	"gorm.io/gorm"
+)
+
+// MessageHandler 是 outbox 消费分发处理器。
+//
+// 设计约束：
+// 1) 入参 envelope 已经完成最外层解析（含 outbox_id、biz_type、payload）；
+// 2) 若返回 nil，表示业务处理成功，框架将继续提交 Kafka offset；
+// 3) 若返回 error，框架会按“可重试错误”处理：回写 outbox 失败状态并进入重试窗口。
+type MessageHandler func(ctx context.Context, envelope kafkabus.Envelope) error
+
+// Engine 是 Outbox + Kafka 的通用异步引擎。
+//
+// 职责边界：
+// 1) 负责 outbox 扫描、Kafka 投递、Kafka 消费与统一状态机流转；
+// 2) 负责 biz_type 到处理器的分发；
+// 3) 不关心具体业务含义（例如“聊天记录落库”），业务语义由 handler 提供。
+//
+// 状态流转口径：
+// pending -> published -> consumed（成功）；
+// pending/published --失败--> pending(带 next_retry_at) 或 dead（达到最大重试）。
+type Engine struct {
+	repo     *Repository
+	producer *kafkabus.Producer
+	consumer *kafkabus.Consumer
+
+	brokers   []string
+	topic     string
+	maxRetry  int
+	scanEvery time.Duration
+	scanBatch int
+
+	handlersMu sync.RWMutex
+	handlers   map[string]MessageHandler
+}
+
+// NewEngine 创建 outbox 异步引擎。
+//
+// 说明：
+// 1) cfg.Enabled=false 时返回 nil，调用方可按“异步关闭”处理；
+// 2) producer/consumer 初始化失败时会确保资源回收，避免半初始化泄漏。
+func NewEngine(repo *Repository, cfg kafkabus.Config) (*Engine, error) {
+	// 1. 配置关闭时直接返回 nil，让上层可以“无侵入降级为同步模式”。
+	if !cfg.Enabled {
+		return nil, nil
+	}
+	// 2. 仓储缺失属于启动期配置错误，直接返回。
+	if repo == nil {
+		return nil, errors.New("outbox repository is nil")
+	}
+
+	// 3. 先初始化 producer，再初始化 consumer。
+	//    如果第二步失败，要主动回收第一步资源，避免泄漏。
+	producer, err := kafkabus.NewProducer(cfg)
+	if err != nil {
+		return nil, err
+	}
+	consumer, err := kafkabus.NewConsumer(cfg)
+	if err != nil {
+		_ = producer.Close()
+		return nil, err
+	}
+
+	// 4. 汇总配置，构造引擎实例。
+	return &Engine{
+		repo:      repo,
+		producer:  producer,
+		consumer:  consumer,
+		brokers:   cfg.Brokers,
+		topic:     cfg.Topic,
+		maxRetry:  cfg.MaxRetry,
+		scanEvery: cfg.RetryScanInterval,
+		scanBatch: cfg.RetryBatchSize,
+		handlers:  make(map[string]MessageHandler),
+	}, nil
+}
+
+// RegisterHandler 注册某个 biz_type 的消费处理器。
+//
+// 设计要求：
+// 1) biz_type 必须唯一，重复注册会覆盖旧值（并打印提示日志）；
+// 2) handler 不能为空；
+// 3) 建议在 Start 前完成注册，减少运行时热更新复杂度。
+func (e *Engine) RegisterHandler(bizType string, handler MessageHandler) error {
+	// 1. 参数校验：防止业务侧在启动链路上把 nil 引擎继续往下用。
+	if e == nil {
+		return errors.New("outbox engine is nil")
+	}
+	// 2. biz_type 为空会导致无法分发，提前拦截。
+	if bizType == "" {
+		return errors.New("bizType is empty")
+	}
+	// 3. handler 为空会在消费时 panic，必须提前拒绝。
+	if handler == nil {
+		return errors.New("handler is nil")
+	}
+
+	// 4. 加写锁更新 handler 映射，保证并发注册时 map 安全。
+	e.handlersMu.Lock()
+	defer e.handlersMu.Unlock()
+	if _, exists := e.handlers[bizType]; exists {
+		log.Printf("outbox handler 覆盖注册: biz_type=%s", bizType)
+	}
+	e.handlers[bizType] = handler
+	return nil
+}
+
+func (e *Engine) getHandler(bizType string) (MessageHandler, bool) {
+	// 读锁足够满足并发读取需求，避免无谓阻塞。
+	e.handlersMu.RLock()
+	defer e.handlersMu.RUnlock()
+	h, ok := e.handlers[bizType]
+	return h, ok
+}
+
+// Start 启动 outbox 异步引擎。
+//
+// 会启动两个后台循环：
+// 1) dispatch loop：扫描 due outbox 并投递到 Kafka；
+// 2) consume loop：消费 Kafka 并按 biz_type 分发处理。
+func (e *Engine) Start(ctx context.Context) {
+	if e == nil {
+		return
+	}
+
+	// 1. 启动日志：把关键运行参数打出来，便于排查“为什么没消费/没扫描”。
+	log.Printf("outbox engine starting: topic=%s brokers=%v retry_scan=%s batch=%d", e.topic, e.brokers, e.scanEvery, e.scanBatch)
+
+	// 2. 启动前探活 topic 是否可用。
+	//    注意：即使探活失败也不会阻断引擎启动，后续循环会继续重试。
+	if err := kafkabus.WaitTopicReady(ctx, e.brokers, e.topic, 30*time.Second); err != nil {
+		log.Printf("Kafka topic not ready before consume loop start: %v", err)
+	} else {
+		log.Printf("Kafka topic is ready: %s", e.topic)
+	}
+
+	// 3. 并行启动两条核心循环：
+	//    - dispatch loop：负责 outbox -> Kafka；
+	//    - consume loop：负责 Kafka -> handler -> outbox 状态推进。
+	go e.startDispatchLoop(ctx)
+	go e.startConsumeLoop(ctx)
+}
+
+// Close 关闭 producer/consumer 资源。
+func (e *Engine) Close() {
+	if e == nil {
+		return
+	}
+	// 逐个关闭并记录错误，避免某个 close 失败导致后续资源无法回收。
+	if err := e.producer.Close(); err != nil {
+		log.Printf("关闭 Kafka producer 失败: %v", err)
+	}
+	if err := e.consumer.Close(); err != nil {
+		log.Printf("关闭 Kafka consumer 失败: %v", err)
+	}
+}
+
+// Enqueue 把业务消息写入 outbox（请求路径调用）。
+//
+// 注意：
+// 1) 该方法不做 Kafka 网络写入，只有数据库写入；
+// 2) messageKey 建议使用业务幂等键（如 conversation_id）以提升分区稳定性；
+// 3) payload 需要可 JSON 序列化。
+func (e *Engine) Enqueue(ctx context.Context, bizType, messageKey string, payload any) error {
+	if e == nil {
+		return errors.New("outbox engine is nil")
+	}
+	// 这里故意只写数据库，不做 Kafka 网络 IO，
+	// 目的是把请求耗时稳定在“单次写库”的可控范围。
+	_, err := e.repo.CreateMessage(ctx, bizType, e.topic, messageKey, payload, e.maxRetry)
+	return err
+}
+
+func (e *Engine) startDispatchLoop(ctx context.Context) {
+	// 1. 定时扫描 due outbox 记录。
+	//    扫描间隔由 scanEvery 控制，避免每次请求都主动触发投递造成抖动。
+	ticker := time.NewTicker(e.scanEvery)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			// 2. 收到退出信号后优雅停止循环。
+			return
+		case <-ticker.C:
+			// 3. 拉取当前窗口内可投递消息。
+			pendingMessages, err := e.repo.ListDueMessages(ctx, e.scanBatch)
+			if err != nil {
+				log.Printf("扫描 outbox 失败: %v", err)
+				continue
+			}
+			if len(pendingMessages) > 0 {
+				log.Printf("outbox due messages=%d, start dispatch", len(pendingMessages))
+			}
+
+			// 4. 逐条投递，单条失败不影响同批后续消息。
+			for _, msg := range pendingMessages {
+				if err = e.dispatchOne(ctx, msg.ID); err != nil {
+					log.Printf("重试投递 outbox 消息失败(id=%d): %v", msg.ID, err)
+				}
+			}
+		}
+	}
+}
+
+func (e *Engine) dispatchOne(ctx context.Context, outboxID int64) error {
+	// 1. 投递前重新按 ID 读取最新状态，避免用到过期快照。
+	outboxMsg, err := e.repo.GetByID(ctx, outboxID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			// 1.1 记录已不存在（可能被清理），按幂等成功处理。
+			return nil
+		}
+		return err
+	}
+	// 1.2 最终态直接跳过，避免重复投递。
+	if outboxMsg.Status == model.OutboxStatusConsumed || outboxMsg.Status == model.OutboxStatusDead {
+		return nil
+	}
+
+	// 2. 组装 Kafka 包装体，统一带上 outbox_id 供消费端做状态回写。
+	envelope := kafkabus.Envelope{
+		OutboxID: outboxMsg.ID,
+		BizType:  outboxMsg.BizType,
+		Payload:  json.RawMessage(outboxMsg.Payload),
+	}
+	raw, err := json.Marshal(envelope)
+	if err != nil {
+		// 2.1 包装层序列化失败通常不可恢复，直接标 dead。
+		markErr := e.repo.MarkDead(ctx, outboxMsg.ID, "序列化 outbox 包装失败: "+err.Error())
+		if markErr != nil {
+			log.Printf("标记 outbox 死信失败(id=%d): %v", outboxMsg.ID, markErr)
+		}
+		return err
+	}
+
+	// 3. 先投 Kafka，再把 outbox 状态推进到 published。
+	//    任一步骤失败都回写 retry，让扫描器后续重试。
+	if err = e.producer.Enqueue(ctx, outboxMsg.Topic, outboxMsg.MessageKey, raw); err != nil {
+		_ = e.repo.MarkFailedForRetry(ctx, outboxMsg.ID, "投递 Kafka 失败: "+err.Error())
+		return err
+	}
+	if err = e.repo.MarkPublished(ctx, outboxMsg.ID); err != nil {
+		_ = e.repo.MarkFailedForRetry(ctx, outboxMsg.ID, "更新已投递状态失败: "+err.Error())
+		return err
+	}
+	return nil
+}
+
+func (e *Engine) startConsumeLoop(ctx context.Context) {
+	// 消费循环采用“拉取 -> 处理 -> 提交 offset”的标准模型。
+	for {
+		select {
+		case <-ctx.Done():
+			// 1. 收到退出信号后终止循环。
+			return
+		default:
+		}
+
+		// 2. 拉取下一条 Kafka 消息。
+		msg, err := e.consumer.Dequeue(ctx)
+		if err != nil {
+			if errors.Is(err, context.Canceled) {
+				// 2.1 context 主动取消时，不记错误日志，直接退出。
+				return
+			}
+			// 2.2 临时错误短暂退避后继续，避免空转刷日志。
+			log.Printf("Kafka 消费拉取失败(topic=%s): %v", e.topic, err)
+			time.Sleep(300 * time.Millisecond)
+			continue
+		}
+
+		// 3. 单条消息处理失败仅记录日志，不阻断消费循环。
+		if err = e.handleMessage(ctx, msg); err != nil {
+			log.Printf("处理 Kafka 消息失败(topic=%s, partition=%d, offset=%d): %v", msg.Topic, msg.Partition, msg.Offset, err)
+		}
+	}
+}
+
+func (e *Engine) handleMessage(ctx context.Context, msg segmentkafka.Message) error {
+	// 1. 先解析最外层 envelope，拿到 outbox_id + biz_type + payload。
+	var envelope kafkabus.Envelope
+	if err := json.Unmarshal(msg.Value, &envelope); err != nil {
+		// 1.1 包装层损坏时无法恢复，直接提交 offset 防止无限重放。
+		_ = e.consumer.Commit(ctx, msg)
+		return fmt.Errorf("解析 Kafka 包装失败: %w", err)
+	}
+	if envelope.OutboxID <= 0 {
+		// 1.2 缺少 outbox_id 无法回写状态，同样提交 offset 跳过。
+		_ = e.consumer.Commit(ctx, msg)
+		return errors.New("Kafka 包装缺少 outbox_id")
+	}
+
+	// 2. 根据 biz_type 查找业务处理器。
+	handler, ok := e.getHandler(envelope.BizType)
+	if !ok {
+		// 2.1 未注册处理器是配置错误，标记 dead 并提交 offset，避免重复消费。
+		_ = e.repo.MarkDead(ctx, envelope.OutboxID, "未知业务类型: "+envelope.BizType)
+		if err := e.consumer.Commit(ctx, msg); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	// 3. 调用业务处理器。
+	if err := handler(ctx, envelope); err != nil {
+		// 统一按“可重试错误”处理，回写 retry 状态后提交 offset，避免同一条消息在 Kafka 侧死循环。
+		if markErr := e.repo.MarkFailedForRetry(ctx, envelope.OutboxID, "消费处理失败: "+err.Error()); markErr != nil {
+			return markErr
+		}
+		if commitErr := e.consumer.Commit(ctx, msg); commitErr != nil {
+			return commitErr
+		}
+		return err
+	}
+
+	// 4. 业务处理成功后提交 offset。
+	return e.consumer.Commit(ctx, msg)
+}
--- a/backend/infra/outbox/repository.go
+++ b/backend/infra/outbox/repository.go
@@ -0,0 +1,259 @@
+package outbox
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/LoveLosita/smartflow/backend/model"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+type Repository struct {
+	db *gorm.DB
+}
+
+// NewRepository 构造 outbox 仓储。
+// 该仓储只关心“数据库状态机”，不关心 Kafka 投递/消费。
+func NewRepository(db *gorm.DB) *Repository {
+	return &Repository{db: db}
+}
+
+// CreateMessage 是通用 outbox 入队入口。
+//
+// 设计说明：
+// 1) 该方法只做“把消息安全写入本地 outbox 表”，不做任何 Kafka 网络调用；
+// 2) next_retry_at 初始化为当前时间，表示“可立即被扫描器捞取”；
+// 3) biz_type 由业务方传入，用于消费侧分发到不同处理器；
+// 4) payload 会被序列化为 JSON 字符串存入 payload 字段，后续再按 biz_type 反序列化。
+//
+// 这也是 Outbox 模式的核心：请求路径只承担本地写库成本，把外部系统不确定性（Kafka 延迟/抖动）
+// 转移给后台异步循环处理。
+func (d *Repository) CreateMessage(ctx context.Context, bizType, topic, messageKey string, payload any, maxRetry int) (int64, error) {
+	// 1. 防御式兜底：若调用方未传 maxRetry，则统一使用默认值 20。
+	//    这样可以避免某些链路遗漏配置导致消息无限重试或零重试。
+	if maxRetry <= 0 {
+		maxRetry = 20
+	}
+
+	// 2. 先把业务载荷序列化成 JSON 字符串。
+	//    序列化失败属于“请求入队前失败”，此时不应创建 outbox 记录，直接返回错误即可。
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return 0, err
+	}
+
+	// 3. 组装 outbox 初始记录：
+	//    - status=pending：表示待投递；
+	//    - retry_count=0：尚未重试；
+	//    - next_retry_at=now：扫描器可立即捞取并尝试首次投递。
+	now := time.Now()
+	msg := model.AgentOutboxMessage{
+		BizType:     bizType,
+		Topic:       topic,
+		MessageKey:  messageKey,
+		Payload:     string(raw),
+		Status:      model.OutboxStatusPending,
+		RetryCount:  0,
+		MaxRetry:    maxRetry,
+		NextRetryAt: &now,
+	}
+
+	// 4. 落库成功后返回 outbox 主键，供上层日志/追踪链路使用。
+	if err = d.db.WithContext(ctx).Create(&msg).Error; err != nil {
+		return 0, err
+	}
+	return msg.ID, nil
+}
+
+// CreateChatHistoryMessage 是聊天记录持久化的兼容入口。
+// 说明：为了避免现有业务调用一次性改太多，先保留该方法作为 CreateMessage 的薄封装。
+func (d *Repository) CreateChatHistoryMessage(ctx context.Context, topic, messageKey string, payload model.ChatHistoryPersistPayload, maxRetry int) (int64, error) {
+	return d.CreateMessage(ctx, model.OutboxBizTypeChatHistoryPersist, topic, messageKey, payload, maxRetry)
+}
+
+// GetByID 按主键读取 outbox 记录。
+// 该方法通常用于 dispatch 前“再读一次最新状态”，避免使用过期快照。
+func (d *Repository) GetByID(ctx context.Context, id int64) (*model.AgentOutboxMessage, error) {
+	var msg model.AgentOutboxMessage
+	if err := d.db.WithContext(ctx).Where("id = ?", id).First(&msg).Error; err != nil {
+		return nil, err
+	}
+	return &msg, nil
+}
+
+// ListDueMessages 拉取“到期可投递”的 pending 消息。
+// 条件说明：
+// 1) status = pending：只处理待投递状态；
+// 2) next_retry_at <= now：到达可重试/可首次投递时间；
+// 3) 按 next_retry_at + id 升序：保证老消息优先，降低饥饿概率。
+func (d *Repository) ListDueMessages(ctx context.Context, limit int) ([]model.AgentOutboxMessage, error) {
+	// 1. 限流兜底，避免误传 0 导致一次拉取过多消息。
+	if limit <= 0 {
+		limit = 100
+	}
+	now := time.Now()
+	var messages []model.AgentOutboxMessage
+	err := d.db.WithContext(ctx).
+		Where("status = ? AND next_retry_at IS NOT NULL AND next_retry_at <= ?", model.OutboxStatusPending, now).
+		Order("next_retry_at ASC, id ASC").
+		Limit(limit).
+		Find(&messages).Error
+	if err != nil {
+		return nil, err
+	}
+	return messages, nil
+}
+
+// MarkPublished 仅在消息未进入最终态时更新为 published，避免覆盖 consumed/dead。
+func (d *Repository) MarkPublished(ctx context.Context, id int64) error {
+	// 1. published 代表“已成功写入 Kafka”。
+	// 2. 清理 last_error/next_retry_at，表示当前无需重试。
+	now := time.Now()
+	updates := map[string]interface{}{
+		"status":        model.OutboxStatusPublished,
+		"published_at":  &now,
+		"last_error":    nil,
+		"next_retry_at": nil,
+	}
+	// 3. 额外加状态保护，避免并发下把 consumed/dead 错误覆盖回 published。
+	result := d.db.WithContext(ctx).
+		Model(&model.AgentOutboxMessage{}).
+		Where("id = ? AND status NOT IN (?, ?)", id, model.OutboxStatusConsumed, model.OutboxStatusDead).
+		Updates(updates)
+	return result.Error
+}
+
+// MarkDead 把消息标记为死信（最终失败，不再重试）。
+// 常见场景：载荷不可反序列化、biz_type 未注册等“不可恢复错误”。
+func (d *Repository) MarkDead(ctx context.Context, id int64, reason string) error {
+	// 1. 错误文本统一裁剪，避免超长错误撑爆字段或日志。
+	now := time.Now()
+	lastErr := truncateError(reason)
+	updates := map[string]interface{}{
+		"status":        model.OutboxStatusDead,
+		"last_error":    &lastErr,
+		"next_retry_at": nil,
+		"updated_at":    now,
+	}
+	return d.db.WithContext(ctx).Model(&model.AgentOutboxMessage{}).Where("id = ?", id).Updates(updates).Error
+}
+
+// MarkFailedForRetry 把一次失败写回 outbox 状态机，并计算下一次重试窗口。
+// 该方法必须在事务内完成“读当前状态 + 写新状态”，保证并发时计数和状态一致。
+func (d *Repository) MarkFailedForRetry(ctx context.Context, id int64, reason string) error {
+	return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		// 1. 行级锁读取，避免多个 goroutine 同时更新同一条消息导致 retry_count 乱序。
+		var msg model.AgentOutboxMessage
+		err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", id).First(&msg).Error
+		if err != nil {
+			return err
+		}
+
+		// 2. 若已是最终态（consumed/dead），直接幂等返回。
+		//    这样即使出现重复调用，也不会把最终态改坏。
+		if msg.Status == model.OutboxStatusConsumed || msg.Status == model.OutboxStatusDead {
+			return nil
+		}
+
+		// 3. 递增重试计数并判断是否达到最大重试次数。
+		nextRetryCount := msg.RetryCount + 1
+		now := time.Now()
+		status := model.OutboxStatusPending
+		var nextRetryAt *time.Time
+		if nextRetryCount >= msg.MaxRetry {
+			// 3.1 达到上限：转 dead，停止后续扫描重试。
+			status = model.OutboxStatusDead
+			nextRetryAt = nil
+		} else {
+			// 3.2 未到上限：按指数退避计算下一次可重试时间。
+			t := now.Add(calcRetryBackoff(nextRetryCount))
+			nextRetryAt = &t
+		}
+
+		// 4. 写回失败原因与状态快照，便于排查问题。
+		lastErr := truncateError(reason)
+		updates := map[string]interface{}{
+			"status":        status,
+			"retry_count":   nextRetryCount,
+			"last_error":    &lastErr,
+			"next_retry_at": nextRetryAt,
+			"updated_at":    now,
+		}
+		return tx.Model(&model.AgentOutboxMessage{}).Where("id = ?", id).Updates(updates).Error
+	})
+}
+
+// PersistChatHistoryAndMarkConsumed 负责“消费成功后落业务库 + 标记 outbox consumed”。
+// 之所以必须放在同一个事务里，是为了保证“业务落库”和“状态推进”原子一致：
+// - 若业务写入失败，不应把 outbox 标记为 consumed；
+// - 若标记 consumed 失败，也应回滚业务写入，避免出现不可追踪的不一致。
+func (d *Repository) PersistChatHistoryAndMarkConsumed(ctx context.Context, outboxID int64, payload model.ChatHistoryPersistPayload) error {
+	return d.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		// 1. 先锁定 outbox 记录，确保同一条消息不会被并发消费者重复推进状态。
+		var outboxMsg model.AgentOutboxMessage
+		err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", outboxID).First(&outboxMsg).Error
+		if err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				// 1.1 幂等兜底：记录不存在时视为“无事可做”。
+				return nil
+			}
+			return err
+		}
+		// 1.2 若已 consumed/dead，说明已被处理过或已终止，直接幂等返回。
+		if outboxMsg.Status == model.OutboxStatusConsumed {
+			return nil
+		}
+		if outboxMsg.Status == model.OutboxStatusDead {
+			return nil
+		}
+
+		// 2. 写入聊天历史业务表（chat_histories）。
+		//    这里不包含 token 统计等扩展字段，只负责核心消息落库。
+		chatMsg := payload.Message
+		chatRole := payload.Role
+		history := model.ChatHistory{
+			UserID:         payload.UserID,
+			ChatID:         payload.ConversationID,
+			MessageContent: &chatMsg,
+			Role:           &chatRole,
+		}
+		if err = tx.Create(&history).Error; err != nil {
+			return err
+		}
+
+		// 3. 业务写入成功后，把 outbox 推进到 consumed 最终态。
+		//    并清理错误与重试字段，表示该消息生命周期结束。
+		now := time.Now()
+		updates := map[string]interface{}{
+			"status":        model.OutboxStatusConsumed,
+			"consumed_at":   &now,
+			"last_error":    nil,
+			"next_retry_at": nil,
+			"updated_at":    now,
+		}
+		return tx.Model(&model.AgentOutboxMessage{}).Where("id = ?", outboxID).Updates(updates).Error
+	})
+}
+
+// calcRetryBackoff 计算指数退避时间。
+// 规则：1s, 2s, 4s, 8s, 16s, 32s（最多封顶到第 6 档）。
+func calcRetryBackoff(retryCount int) time.Duration {
+	if retryCount <= 0 {
+		return time.Second
+	}
+	if retryCount > 6 {
+		retryCount = 6
+	}
+	return time.Second * time.Duration(1<<(retryCount-1))
+}
+
+// truncateError 限制错误文本最大长度，防止写库失败或日志污染。
+func truncateError(reason string) string {
+	if len(reason) <= 2000 {
+		return reason
+	}
+	return reason[:2000]
+}