Version: 0.9.61.dev.260501

后端:
1. 主动调度 graph + session bridge 收口——把 dry-run / select / preview / confirm / rerun 串成受限 graph,新增 active_schedule_sessions 缓存与聊天拦截,ready_preview 后释放回自由聊天
2. 会话与通知链路对齐——notification 统一绑定 conversation_id,action_url 指向 /assistant/{conversation_id},会话不存在改回 404 语义,避免 wrong param type 误导排障
3. estimated_sections 写入与主动调度消费链路补齐——任务创建、quick task 与随口记入口都透传估计节数,主动调度只消费落库值

前端:
4. AssistantPanel 最小适配主动调度预览与失败态——复用主动调度卡片/微调弹窗,补历史加载失败可见提示与跨账号会话拦截

文档:
5. 更新主动调度缺口分阶段实施计划和实现方案,标记阶段 0-2 收口并同步接力状态
This commit is contained in:
Losita
2026-05-01 20:48:32 +08:00
parent 0a014f7472
commit a3eaa9b2c2
42 changed files with 4377 additions and 357 deletions

View File

@@ -0,0 +1,198 @@
package graph
import (
"context"
"errors"
"fmt"
"github.com/LoveLosita/smartflow/backend/active_scheduler/candidate"
schedulercontext "github.com/LoveLosita/smartflow/backend/active_scheduler/context"
"github.com/LoveLosita/smartflow/backend/active_scheduler/observe"
"github.com/LoveLosita/smartflow/backend/active_scheduler/selection"
"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
"github.com/cloudwego/eino/compose"
)
const GraphName = "active_schedule_graph"
const (
NodeDryRun = "dry_run"
NodeSelect = "select"
)
// DryRunData 承载 graph 中 dry-run 节点产出的只读结果。
//
// 职责边界:
// 1. 只装载构造 preview / 选择器所需的 dry-run 结果;
// 2. 不包含落库状态,也不包含通知副作用;
// 3. 由 graph runner 负责串联后续选择步骤。
type DryRunData struct {
Context *schedulercontext.ActiveScheduleContext
Observation observe.Result
Candidates []candidate.Candidate
}
// DryRunFunc 描述 graph 依赖的 dry-run 执行入口。
//
// 职责边界:
// 1. 只负责把 trigger 变成 dry-run 结果;
// 2. 不负责 selection / preview / notification
// 3. 由 service 层用现有 DryRunService 做适配注入。
type DryRunFunc func(ctx context.Context, trig trigger.ActiveScheduleTrigger) (*DryRunData, error)
// Selector 描述 graph 依赖的候选选择器。
//
// 职责边界:
// 1. 只负责在后端候选里做有限选择与解释生成;
// 2. 不负责构造 dry-run 结果;
// 3. 不负责 preview 落库与通知投递。
type Selector interface {
Select(ctx context.Context, req selection.SelectRequest) (selection.Result, error)
}
// RunResult 是 graph 的最终输出。
//
// 职责边界:
// 1. 只回传 dry-run 与选择结果;
// 2. 不包含 preview / notification 的持久化结果;
// 3. 上层 service 仍负责事务、落库和 outbox。
type RunResult struct {
DryRunData *DryRunData
SelectionResult selection.Result
}
type runState struct {
Trigger trigger.ActiveScheduleTrigger
DryRunData *DryRunData
SelectionResult selection.Result
}
// Runner 把 dry-run 与受限选择串成一条可复用的 graph。
type Runner struct {
dryRun DryRunFunc
selector Selector
}
// NewRunner 创建主动调度 graph runner。
func NewRunner(dryRun DryRunFunc, selector Selector) (*Runner, error) {
if dryRun == nil {
return nil, errors.New("active scheduler dry-run 不能为空")
}
if selector == nil {
return nil, errors.New("active scheduler selector 不能为空")
}
return &Runner{dryRun: dryRun, selector: selector}, nil
}
// Run 执行 dry-run -> select 的 graph。
//
// 步骤化说明:
// 1. 先跑 dry-run拿到上下文、观测结果与候选
// 2. 再交给受限选择器做选中与解释生成;
// 3. 任一步失败都直接返回,避免写入半截 preview
// 4. 上层 service 再决定是否写 preview / 发通知。
func (r *Runner) Run(ctx context.Context, trig trigger.ActiveScheduleTrigger) (*RunResult, error) {
if r == nil || r.dryRun == nil || r.selector == nil {
return nil, errors.New("active scheduler graph runner 未初始化")
}
state := &runState{Trigger: trig}
g := compose.NewGraph[*runState, *runState]()
if err := g.AddLambdaNode(NodeDryRun, compose.InvokableLambda(r.dryRunNode())); err != nil {
return nil, err
}
if err := g.AddLambdaNode(NodeSelect, compose.InvokableLambda(r.selectNode())); err != nil {
return nil, err
}
if err := g.AddEdge(compose.START, NodeDryRun); err != nil {
return nil, err
}
if err := g.AddEdge(NodeDryRun, NodeSelect); err != nil {
return nil, err
}
if err := g.AddEdge(NodeSelect, compose.END); err != nil {
return nil, err
}
runnable, err := g.Compile(ctx,
compose.WithGraphName(GraphName),
compose.WithMaxRunSteps(8),
compose.WithNodeTriggerMode(compose.AnyPredecessor),
)
if err != nil {
return nil, err
}
out, err := runnable.Invoke(ctx, state)
if err != nil {
return nil, err
}
if out == nil {
return nil, errors.New("active scheduler graph 返回空状态")
}
return &RunResult{
DryRunData: out.DryRunData,
SelectionResult: out.SelectionResult,
}, nil
}
func (r *Runner) dryRunNode() func(context.Context, *runState) (*runState, error) {
return func(ctx context.Context, state *runState) (*runState, error) {
if state == nil {
return nil, errors.New("active scheduler graph state 不能为空")
}
if r == nil || r.dryRun == nil {
return nil, errors.New("active scheduler dry-run 不能为空")
}
// 1. 先跑 dry-run汇集上下文、观测结果和候选避免 selection 直接依赖数据库。
// 2. dry-run 出错时直接返回,不进入后续选择。
result, err := r.dryRun(ctx, state.Trigger)
if err != nil {
return nil, err
}
state.DryRunData = result
return state, nil
}
}
func (r *Runner) selectNode() func(context.Context, *runState) (*runState, error) {
return func(ctx context.Context, state *runState) (*runState, error) {
if state == nil {
return nil, errors.New("active scheduler graph state 不能为空")
}
if state.DryRunData == nil {
return nil, errors.New("active scheduler graph 缺少 dry-run 结果")
}
if r == nil || r.selector == nil {
return nil, errors.New("active scheduler selector 不能为空")
}
// 1. 没有候选时,不强行调用选择器,直接返回空选择结果,交给上层决定是否继续写 preview。
// 2. 有候选时再做受限 LLM 选择,确保模型只看后端生成的候选视图。
if len(state.DryRunData.Candidates) == 0 {
state.SelectionResult = selection.Result{}
return state, nil
}
result, err := r.selector.Select(ctx, selection.SelectRequest{
ActiveContext: state.DryRunData.Context,
Observation: state.DryRunData.Observation,
Candidates: state.DryRunData.Candidates,
})
if err != nil {
return nil, err
}
state.SelectionResult = result
return state, nil
}
}
func (r *Runner) String() string {
if r == nil {
return "active_scheduler_graph(nil)"
}
return fmt.Sprintf("active_scheduler_graph(dry_run=%t, selector=%t)", r.dryRun != nil, r.selector != nil)
}

View File

@@ -174,7 +174,7 @@ func entryFromEvent(event ports.ScheduleEventFact) SchedulePreviewEntry {
return entry
}
func riskDTO(selected candidate.Candidate, observation observe.Result, changes []ActiveScheduleChangeItem) RiskDTO {
func riskDTO(selected candidate.Candidate, observation observe.Result, changes []ActiveScheduleChangeItem, fallbackUsed bool) RiskDTO {
affectedIDs := make([]int, 0)
seen := make(map[int]bool)
for _, change := range changes {
@@ -198,7 +198,7 @@ func riskDTO(selected candidate.Candidate, observation observe.Result, changes [
RiskMetrics: observation.Metrics.Risk,
AffectedIDs: affectedIDs,
RequiresLLM: observation.Decision.LLMSelectionRequired,
FallbackUsed: observation.Decision.FallbackCandidateID == selected.CandidateID,
FallbackUsed: fallbackUsed,
}
}

View File

@@ -22,10 +22,12 @@ type CreatePreviewRequest struct {
Candidates []candidate.Candidate `json:"-"`
PreviewID string `json:"preview_id,omitempty"`
TriggerID string `json:"trigger_id,omitempty"`
SelectedCandidateID string `json:"selected_candidate_id,omitempty"`
BaseVersion string `json:"base_version,omitempty"`
GeneratedAt time.Time `json:"generated_at,omitempty"`
ExplanationText string `json:"explanation_text,omitempty"`
NotificationSummary string `json:"notification_summary,omitempty"`
FallbackUsed bool `json:"fallback_used,omitempty"`
}
// CreatePreviewResponse 是写入 preview 后可直接返回给 API 的响应 DTO。

View File

@@ -65,9 +65,10 @@ func (s *Service) SetClock(clock func() time.Time) {
// CreatePreview 把 dry-run 结果保存为 ready preview。
//
// 职责边界:
// 1. 只消费已经完成的 dry-run 结果,不重新读取任务/日程事实;
// 2. MVP 没有 LLM 选择器,固定使用后端排序后的 top1 candidate 作为 selected_candidate
// 3. 写库后只返回详情 DTO不发布通知、不正式应用候选、不回写 trigger。
// 1. 只消费已经完成的 dry-run 结果,不重新读取任务/日程事实;
// 2. 优先吃上层 selection 结果中的 selected_candidate_id / explanation / notification 摘要
// 若上层未显式传入,则为了兼容旧链路继续回退到 top1 candidate
// 3. 写库后只返回详情 DTO不发布通知、不正式应用候选、不回写 trigger。
func (s *Service) CreatePreview(ctx context.Context, req CreatePreviewRequest) (*CreatePreviewResponse, error) {
if s == nil || s.repo == nil {
return nil, fmt.Errorf("%w: preview service 未初始化", ErrInvalidPreviewRequest)
@@ -97,9 +98,15 @@ func (s *Service) CreatePreview(ctx context.Context, req CreatePreviewRequest) (
previewID = "asp_" + uuid.NewString()
}
// 1. 先构造所有展示快照,再写库;任何 JSON 转换失败都提前返回,避免落入半结构化记录。
selected := req.Candidates[0]
snapshot := buildSnapshot(activeContext, req.Observation, req.Candidates, selected)
// 1. 先解析选中的候选,再构造展示快照;任何 JSON 转换失败都提前返回,避免落入半结构化记录。
// 1.1 若上层已经给出 selected_candidate_id就严格按该候选落库避免 preview 与选择结果不一致。
// 1.2 若未给出,则继续沿用后端候选顺序的第一条,保持旧流程兼容。
// 1.3 若指定 ID 不在候选列表中,直接返回错误,避免写入一份错位的 preview。
selected, err := pickSelectedCandidate(req.Candidates, req.SelectedCandidateID)
if err != nil {
return nil, err
}
snapshot := buildSnapshot(activeContext, req.Observation, req.Candidates, selected, req.FallbackUsed)
baseVersion := strings.TrimSpace(req.BaseVersion)
if baseVersion == "" {
baseVersion = buildBaseVersion(activeContext, snapshot.changes)
@@ -170,6 +177,24 @@ func (s *Service) now() time.Time {
return s.clock()
}
func pickSelectedCandidate(candidates []candidate.Candidate, selectedCandidateID string) (candidate.Candidate, error) {
if len(candidates) == 0 {
return candidate.Candidate{}, fmt.Errorf("%w: dry-run 链路未生成可保存候选", ErrInvalidPreviewRequest)
}
selectedCandidateID = strings.TrimSpace(selectedCandidateID)
if selectedCandidateID == "" {
return candidates[0], nil
}
for _, item := range candidates {
if strings.TrimSpace(item.CandidateID) == selectedCandidateID {
return item, nil
}
}
return candidate.Candidate{}, fmt.Errorf("%w: selected_candidate_id 不在候选列表中", ErrInvalidPreviewRequest)
}
func buildPreviewModel(
previewID string,
triggerID string,
@@ -256,6 +281,7 @@ func buildSnapshot(
observation observe.Result,
candidates []candidate.Candidate,
selected candidate.Candidate,
fallbackUsed bool,
) rawPreviewSnapshot {
selectedDTO := candidateDTO(selected)
candidateDTOs := make([]CandidateDTO, 0, len(candidates))
@@ -276,6 +302,6 @@ func buildSnapshot(
before: before,
changes: changes,
after: after,
risk: riskDTO(selected, observation, changes),
risk: riskDTO(selected, observation, changes, fallbackUsed),
}
}

View File

@@ -0,0 +1,61 @@
package selection
import (
"github.com/LoveLosita/smartflow/backend/active_scheduler/candidate"
schedulercontext "github.com/LoveLosita/smartflow/backend/active_scheduler/context"
"github.com/LoveLosita/smartflow/backend/active_scheduler/observe"
)
const (
ActionSelectCandidate = "select_candidate"
ActionAskUser = "ask_user"
ActionNotifyOnly = "notify_only"
ActionClose = "close"
)
// SelectRequest 是主动调度选择器的输入。
//
// 职责边界:
// 1. 只承载已经由 dry-run 生成并校验过的上下文、观测结果和候选;
// 2. 不包含任何模型实例,不负责 prompt 拼接;
// 3. 由 graph runner 在 dry-run 之后传入,避免选择器直接回查数据库。
type SelectRequest struct {
ActiveContext *schedulercontext.ActiveScheduleContext `json:"-"`
Observation observe.Result `json:"-"`
Candidates []candidate.Candidate `json:"-"`
}
// Result 是选择器的结构化输出。
//
// 职责边界:
// 1. 只记录最终选中的候选与给用户看的解释摘要;
// 2. 不包含正式日程写入结果,也不包含通知投递结果;
// 3. FallbackUsed 只表示本次是否回退到了确定性兜底,不允许靠 selected_candidate_id 推断。
type Result struct {
Action string `json:"action"`
SelectedCandidateID string `json:"selected_candidate_id,omitempty"`
Reason string `json:"reason,omitempty"`
ExplanationText string `json:"explanation_text,omitempty"`
NotificationSummary string `json:"notification_summary,omitempty"`
AskUserQuestion string `json:"ask_user_question,omitempty"`
FallbackUsed bool `json:"fallback_used,omitempty"`
Confidence float64 `json:"confidence,omitempty"`
}
// CandidateView 是暴露给 LLM 的最小候选视图。
//
// 职责边界:
// 1. 只保留用于有限选择的基础信息和少量结构化维度;
// 2. 不暴露 score / validation 这类内部实现细节;
// 3. 不直接携带原始日程事实,避免模型看到过多上下文。
type CandidateView struct {
CandidateID string `json:"candidate_id"`
CandidateType string `json:"candidate_type"`
Title string `json:"title"`
Summary string `json:"summary"`
BeforeSummary string `json:"before_summary"`
AfterSummary string `json:"after_summary"`
ChangeSummary string `json:"change_summary"`
CapacityFit string `json:"capacity_fit"`
RiskLevel string `json:"risk_level"`
}

View File

@@ -0,0 +1,232 @@
package selection
import (
"encoding/json"
"fmt"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/active_scheduler/candidate"
)
const selectionSystemPrompt = `
你是 SmartFlow 主动调度的候选选择器。
你的职责很窄:
1. 只在后端已经生成并校验过的候选里做有限选择;
2. 只参考结构化事实、候选基础信息、capacity_fit 和 risk_level
3. 不要输出推理过程,不要输出 score不要输出 confidence不要编造新的候选
4. 只输出 JSON不要输出 markdown不要输出解释性正文。
允许的 action
- select_candidate
- ask_user
- notify_only
- close
输出 JSON 结构:
{
"action": "select_candidate",
"selected_candidate_id": "cand_xxx",
"reason": "简短选择理由",
"explanation_text": "给用户看的简短解释",
"notification_summary": "通知里要显示的简短摘要",
"ask_user_question": "需要追问时填写,否则留空"
}
规则:
1. selected_candidate_id 必须来自候选列表;如果 action=close也要选候选列表里对应的 close 候选;
2. 如果需要追问,优先选择 ask_user 候选,并把 ask_user_question 写清楚;
3. 如果只需要提醒,不要编造正式日程变更;
4. 如果候选里已经有明显更稳妥的项,优先选风险更低且 capacity_fit 更好的那个;
5. 不要试图重排整段日程,第一版只在候选之间做有限裁决;
6. 如果信息不足,就直接走 ask_user不要硬猜。
`
type selectionPromptInput struct {
GeneratedAt string `json:"generated_at"`
Trigger selectionTriggerInput `json:"trigger"`
DecisionHint selectionDecisionInput `json:"decision_hint"`
Context selectionContextInput `json:"context"`
Candidates []CandidateView `json:"candidates"`
}
type selectionTriggerInput struct {
TriggerID string `json:"trigger_id"`
TriggerType string `json:"trigger_type"`
Source string `json:"source"`
TargetType string `json:"target_type"`
TargetID int `json:"target_id"`
TargetTitle string `json:"target_title"`
RequestedAt string `json:"requested_at"`
TraceID string `json:"trace_id"`
}
type selectionDecisionInput struct {
Action string `json:"action"`
PrimaryIssueCode string `json:"primary_issue_code"`
ReasonCode string `json:"reason_code"`
ShouldNotify bool `json:"should_notify"`
ShouldWritePreview bool `json:"should_write_preview"`
FallbackCandidateID string `json:"fallback_candidate_id,omitempty"`
}
type selectionContextInput struct {
WindowStart string `json:"window_start"`
WindowEnd string `json:"window_end"`
WindowReason string `json:"window_reason"`
MissingInfo []string `json:"missing_info"`
Warnings []string `json:"warnings"`
TraceSteps []string `json:"trace_steps"`
}
func buildSelectionPromptInput(req SelectRequest, now time.Time) selectionPromptInput {
activeContext := req.ActiveContext
decision := req.Observation.Decision
input := selectionPromptInput{
GeneratedAt: now.In(time.Local).Format(time.RFC3339),
DecisionHint: selectionDecisionInput{
Action: string(decision.Action),
PrimaryIssueCode: string(decision.PrimaryIssueCode),
ReasonCode: decision.ReasonCode,
ShouldNotify: decision.ShouldNotify,
ShouldWritePreview: decision.ShouldWritePreview,
FallbackCandidateID: strings.TrimSpace(decision.FallbackCandidateID),
},
}
if activeContext != nil {
input.Trigger = selectionTriggerInput{
TriggerID: activeContext.Trigger.TriggerID,
TriggerType: string(activeContext.Trigger.TriggerType),
Source: string(activeContext.Trigger.Source),
TargetType: string(activeContext.Trigger.TargetType),
TargetID: activeContext.Trigger.TargetID,
TargetTitle: activeContext.Target.Title,
RequestedAt: activeContext.Trigger.RequestedAt.In(time.Local).Format(time.RFC3339),
TraceID: activeContext.Trace.TraceID,
}
input.Context = selectionContextInput{
WindowStart: activeContext.Window.StartAt.In(time.Local).Format(time.RFC3339),
WindowEnd: activeContext.Window.EndAt.In(time.Local).Format(time.RFC3339),
WindowReason: activeContext.Window.WindowReason,
MissingInfo: append([]string(nil), activeContext.DerivedFacts.MissingInfo...),
Warnings: append([]string(nil), activeContext.Trace.Warnings...),
TraceSteps: append([]string(nil), activeContext.Trace.BuildSteps...),
}
}
input.Candidates = make([]CandidateView, 0, len(req.Candidates))
for _, item := range req.Candidates {
input.Candidates = append(input.Candidates, buildCandidateView(req, item))
}
return input
}
func buildSelectionUserPrompt(input selectionPromptInput) (string, error) {
raw, err := json.MarshalIndent(input, "", " ")
if err != nil {
return "", err
}
var sb strings.Builder
sb.WriteString("请基于下面的结构化事实,从候选中选一个最合适的结果。\n")
sb.WriteString("输入:\n")
sb.WriteString(string(raw))
sb.WriteString("\n")
sb.WriteString("只输出 JSON。")
return sb.String(), nil
}
func buildCandidateView(req SelectRequest, item candidate.Candidate) CandidateView {
return CandidateView{
CandidateID: item.CandidateID,
CandidateType: string(item.CandidateType),
Title: item.Title,
Summary: item.Summary,
BeforeSummary: item.BeforeSummary,
AfterSummary: item.AfterSummary,
ChangeSummary: buildChangeSummary(item),
CapacityFit: deriveCapacityFit(req, item),
RiskLevel: deriveRiskLevel(req, item),
}
}
func buildChangeSummary(item candidate.Candidate) string {
if len(item.Changes) == 0 {
return "无正式变更"
}
lines := make([]string, 0, len(item.Changes))
for _, change := range item.Changes {
lines = append(lines, summarizeChange(change))
}
return strings.Join(lines, "")
}
func summarizeChange(change candidate.ChangeItem) string {
switch change.ChangeType {
case candidate.ChangeTypeAdd:
if change.ToSlot != nil {
return fmt.Sprintf("新增到 第%d周 第%d天 第%d-%d节持续%d节",
change.ToSlot.Start.Week,
change.ToSlot.Start.DayOfWeek,
change.ToSlot.Start.Section,
change.ToSlot.End.Section,
change.DurationSections,
)
}
return "新增一段日程"
case candidate.ChangeTypeCreateMakeup:
if change.ToSlot != nil {
return fmt.Sprintf("为目标补做一段第%d周第%d天第%d-%d节的时间块",
change.ToSlot.Start.Week,
change.ToSlot.Start.DayOfWeek,
change.ToSlot.Start.Section,
change.ToSlot.End.Section,
)
}
return "新增补做块"
case candidate.ChangeTypeAskUser:
return "需要用户补充信息"
default:
return "不修改正式日程"
}
}
func deriveCapacityFit(req SelectRequest, item candidate.Candidate) string {
switch item.CandidateType {
case candidate.TypeAskUser, candidate.TypeNotifyOnly, candidate.TypeClose:
return "not_applicable"
}
if !item.Validation.Valid {
return "insufficient"
}
gap := req.Observation.Metrics.Window.CapacityGap
switch {
case gap > 0:
return "insufficient"
case gap == 0:
return "tight"
default:
return "fit"
}
}
func deriveRiskLevel(req SelectRequest, item candidate.Candidate) string {
if !item.Validation.Valid {
return "high"
}
switch item.CandidateType {
case candidate.TypeCreateMakeup:
return "medium"
case candidate.TypeAddTaskPoolToSchedule:
if req.Observation.Metrics.Window.CapacityGap == 0 {
return "medium"
}
return "low"
case candidate.TypeAskUser, candidate.TypeNotifyOnly, candidate.TypeClose:
return "low"
default:
return "low"
}
}

View File

@@ -0,0 +1,303 @@
package selection
import (
"context"
"errors"
"fmt"
"log"
"math"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/active_scheduler/candidate"
infrallm "github.com/LoveLosita/smartflow/backend/infra/llm"
)
const selectionMaxTokens = 1200
// Service 负责主动调度候选的受限 LLM 选择。
//
// 职责边界:
// 1. 只在后端候选中做选择和解释生成,不生成新候选;
// 2. LLM 失败、输出非法或选择不存在候选时,回退到后端 fallback candidate
// 3. 不写 preview、不发通知、不修改正式日程。
type Service struct {
client *infrallm.Client
clock func() time.Time
logger *log.Logger
}
// NewService 创建主动调度选择器。
//
// 说明:
// 1. client 允许为空;为空时选择器只走确定性 fallback便于本地测试和降级
// 2. 真正的模型接入在 cmd/start.go 中完成aiHub.Pro -> llm.Client -> selection.Service
// 3. 选择器本身不持有模型配置,只表达本业务域的 prompt 和结果校验。
func NewService(client *infrallm.Client) *Service {
return &Service{
client: client,
clock: time.Now,
logger: log.Default(),
}
}
func (s *Service) SetClock(clock func() time.Time) {
if s != nil && clock != nil {
s.clock = clock
}
}
// Select 对主动调度候选做有限选择。
//
// 步骤化说明:
// 1. 先校验 dry-run 输入,保证 LLM 不会拿到空上下文或空候选;
// 2. 若模型不可用,直接使用后端 fallback candidate并显式标记 FallbackUsed
// 3. 若模型可用,构造只读候选视图,隐藏 score / confidence / 原始事实快照;
// 4. 校验 LLM 返回的 candidate_id 是否存在,非法则回退;
// 5. 最终结果只交给 preview 层落库,不在这里产生任何副作用。
func (s *Service) Select(ctx context.Context, req SelectRequest) (Result, error) {
if err := validateRequest(req); err != nil {
return Result{}, err
}
if s == nil || s.client == nil {
return buildFallbackResult(req, "模型客户端未配置"), nil
}
input := buildSelectionPromptInput(req, s.now())
userPrompt, err := buildSelectionUserPrompt(input)
if err != nil {
return buildFallbackResult(req, "选择器 prompt 构造失败: "+err.Error()), nil
}
messages := infrallm.BuildSystemUserMessages(
strings.TrimSpace(selectionSystemPrompt),
nil,
userPrompt,
)
resp, rawResult, err := infrallm.GenerateJSON[llmSelectionResponse](
ctx,
s.client,
messages,
infrallm.GenerateOptions{
Temperature: 0.1,
MaxTokens: selectionMaxTokens,
Thinking: infrallm.ThinkingModeDisabled,
Metadata: map[string]any{
"stage": "active_scheduler_select",
"candidate_count": len(req.Candidates),
},
},
)
if err != nil {
if s.logger != nil {
s.logger.Printf("[WARN] 主动调度 LLM 选择失败,使用 fallback: err=%v raw=%s", err, truncateRaw(rawResult))
}
return buildFallbackResult(req, "模型选择失败: "+err.Error()), nil
}
result, fallbackUsed := convertLLMResponse(req, resp)
if fallbackUsed && s.logger != nil {
selectedCandidateID := ""
action := ""
if resp != nil {
selectedCandidateID = strings.TrimSpace(resp.SelectedCandidateID)
action = strings.TrimSpace(resp.Action)
}
s.logger.Printf("[WARN] 主动调度 LLM 选择结果非法,使用 fallback: selected=%q action=%q",
selectedCandidateID,
action,
)
}
return result, nil
}
type llmSelectionResponse struct {
Action string `json:"action"`
SelectedCandidateID string `json:"selected_candidate_id"`
Reason string `json:"reason"`
ExplanationText string `json:"explanation_text"`
NotificationSummary string `json:"notification_summary"`
AskUserQuestion string `json:"ask_user_question"`
}
func validateRequest(req SelectRequest) error {
if req.ActiveContext == nil {
return errors.New("active scheduler selection 缺少上下文")
}
if len(req.Candidates) == 0 {
return errors.New("active scheduler selection 缺少候选")
}
return nil
}
func convertLLMResponse(req SelectRequest, resp *llmSelectionResponse) (Result, bool) {
if resp == nil {
return buildFallbackResult(req, "模型返回空选择结果"), true
}
selected, ok := findCandidate(req.Candidates, resp.SelectedCandidateID)
if !ok {
return buildFallbackResult(req, "模型选择了不存在的候选"), true
}
inferredAction := inferAction(selected)
action := normalizeAction(resp.Action)
fallbackUsed := false
if action == "" || !isActionCompatible(action, selected.CandidateType) {
action = inferredAction
fallbackUsed = true
}
explanation := firstNonEmpty(resp.ExplanationText, selected.Summary)
notificationSummary := firstNonEmpty(resp.NotificationSummary, explanation, selected.Summary)
askUserQuestion := strings.TrimSpace(resp.AskUserQuestion)
if action == ActionAskUser && askUserQuestion == "" {
askUserQuestion = explanation
}
return Result{
Action: action,
SelectedCandidateID: selected.CandidateID,
Reason: strings.TrimSpace(resp.Reason),
ExplanationText: explanation,
NotificationSummary: notificationSummary,
AskUserQuestion: askUserQuestion,
FallbackUsed: fallbackUsed,
Confidence: deriveInternalConfidence(selected),
}, fallbackUsed
}
func buildFallbackResult(req SelectRequest, reason string) Result {
selected := pickFallbackCandidate(req)
action := inferAction(selected)
explanation := firstNonEmpty(selected.Summary, reason)
return Result{
Action: action,
SelectedCandidateID: selected.CandidateID,
Reason: strings.TrimSpace(reason),
ExplanationText: explanation,
NotificationSummary: explanation,
AskUserQuestion: fallbackAskUserQuestion(action, explanation),
FallbackUsed: true,
Confidence: deriveInternalConfidence(selected),
}
}
func pickFallbackCandidate(req SelectRequest) candidate.Candidate {
fallbackID := strings.TrimSpace(req.Observation.Decision.FallbackCandidateID)
if fallbackID != "" {
if selected, ok := findCandidate(req.Candidates, fallbackID); ok {
return selected
}
}
return req.Candidates[0]
}
func findCandidate(candidates []candidate.Candidate, id string) (candidate.Candidate, bool) {
id = strings.TrimSpace(id)
if id == "" {
return candidate.Candidate{}, false
}
for _, item := range candidates {
if strings.TrimSpace(item.CandidateID) == id {
return item, true
}
}
return candidate.Candidate{}, false
}
func normalizeAction(raw string) string {
switch strings.ToLower(strings.TrimSpace(raw)) {
case ActionSelectCandidate:
return ActionSelectCandidate
case ActionAskUser:
return ActionAskUser
case ActionNotifyOnly:
return ActionNotifyOnly
case ActionClose:
return ActionClose
default:
return ""
}
}
func inferAction(item candidate.Candidate) string {
switch item.CandidateType {
case candidate.TypeAskUser:
return ActionAskUser
case candidate.TypeNotifyOnly:
return ActionNotifyOnly
case candidate.TypeClose:
return ActionClose
default:
return ActionSelectCandidate
}
}
func isActionCompatible(action string, candidateType candidate.Type) bool {
switch candidateType {
case candidate.TypeAskUser:
return action == ActionAskUser
case candidate.TypeNotifyOnly:
return action == ActionNotifyOnly
case candidate.TypeClose:
return action == ActionClose
default:
return action == ActionSelectCandidate
}
}
func fallbackAskUserQuestion(action string, explanation string) string {
if action != ActionAskUser {
return ""
}
return strings.TrimSpace(explanation)
}
func deriveInternalConfidence(item candidate.Candidate) float64 {
if !item.Validation.Valid {
return 0.2
}
if item.Score <= 0 {
return 0.55
}
score := float64(item.Score) / 100
return math.Max(0.35, math.Min(0.95, score))
}
func firstNonEmpty(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
func truncateRaw(raw *infrallm.TextResult) string {
if raw == nil {
return ""
}
text := strings.TrimSpace(raw.Text)
runes := []rune(text)
if len(runes) <= 200 {
return text
}
return string(runes[:200]) + "..."
}
func (s *Service) now() time.Time {
if s == nil || s.clock == nil {
return time.Now()
}
return s.clock()
}
func (r Result) String() string {
return fmt.Sprintf("active_scheduler_selection(action=%s, selected=%s, fallback=%t)",
r.Action,
r.SelectedCandidateID,
r.FallbackUsed,
)
}

View File

@@ -0,0 +1,31 @@
package service
import (
"context"
activegraph "github.com/LoveLosita/smartflow/backend/active_scheduler/graph"
"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
)
// AsGraphDryRunFunc 把现有 dry-run service 适配成 graph runner 可用的入口。
//
// 职责边界:
// 1. 只做 service.Result -> graph.DryRunData 的轻量转换;
// 2. 不改写 dry-run 行为,不引入额外候选逻辑;
// 3. 让 graph runner 可以复用现有 BuildContext -> Observe -> GenerateCandidates 链路。
func (s *DryRunService) AsGraphDryRunFunc() activegraph.DryRunFunc {
if s == nil {
return nil
}
return func(ctx context.Context, trig trigger.ActiveScheduleTrigger) (*activegraph.DryRunData, error) {
result, err := s.DryRun(ctx, trig)
if err != nil {
return nil, err
}
return &activegraph.DryRunData{
Context: result.Context,
Observation: result.Observation,
Candidates: result.Candidates,
}, nil
}
}

View File

@@ -0,0 +1,302 @@
package service
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
activepreview "github.com/LoveLosita/smartflow/backend/active_scheduler/preview"
"github.com/LoveLosita/smartflow/backend/active_scheduler/selection"
"github.com/LoveLosita/smartflow/backend/dao"
"github.com/LoveLosita/smartflow/backend/model"
"github.com/google/uuid"
"gorm.io/gorm"
)
var (
activeScheduleConversationNamespace = uuid.NewSHA1(uuid.NameSpaceURL, []byte("smartflow:active_schedule:conversation"))
activeScheduleSessionNamespace = uuid.NewSHA1(uuid.NameSpaceURL, []byte("smartflow:active_schedule:session"))
)
// WithActiveScheduleSessionBridge 注入主动调度 session 预创建所需的 DAO。
//
// 职责边界:
// 1. 只把 trigger -> notification 前的会话桥接能力接入 workflow
// 2. 不改变 dry-run / preview / notification 的主状态机;
// 3. 为空时保留旧能力,便于局部测试与迁移期回退。
func WithActiveScheduleSessionBridge(agentDAO *dao.AgentDAO, sessionDAO *dao.ActiveScheduleSessionDAO) TriggerWorkflowOption {
return func(s *TriggerWorkflowService) {
if s == nil {
return
}
s.agentDAO = agentDAO
s.sessionDAO = sessionDAO
}
}
// bootstrapActiveScheduleConversationInTx 负责在 notification 发出前预建会话与首屏内容。
//
// 步骤化说明:
// 1. 先生成确定性的 conversation/session ID保证 trigger 重试时不会拆成多条会话;
// 2. 再在同一事务里创建或复用 agent_chats 和 active_schedule_sessions
// 3. 如果是首次落库,则顺手补一条 assistant_text必要时再补一张主动调度卡片
// 4. 任一步失败都直接返回 error让上层事务整体回滚避免“通知已发但会话底稿没落”。
func (s *TriggerWorkflowService) bootstrapActiveScheduleConversationInTx(
ctx context.Context,
tx *gorm.DB,
triggerRow model.ActiveScheduleTrigger,
previewDetail activepreview.ActiveSchedulePreviewDetail,
selectionResult selection.Result,
now time.Time,
) error {
if s == nil {
return errors.New("主动调度会话桥未初始化")
}
if s.agentDAO == nil || s.sessionDAO == nil {
return nil
}
if tx == nil {
return errors.New("gorm tx 不能为空")
}
if triggerRow.ID == "" {
return errors.New("trigger_id 不能为空")
}
conversationID := buildActiveScheduleConversationID(triggerRow.ID)
sessionID := buildActiveScheduleSessionID(triggerRow.ID)
txAgentDAO := s.agentDAO.WithTx(tx)
txSessionDAO := s.sessionDAO.WithTx(tx)
if err := ensureAgentConversationExists(ctx, txAgentDAO, triggerRow.UserID, conversationID); err != nil {
return err
}
baseSeq, err := txAgentDAO.GetConversationTimelineMaxSeq(ctx, triggerRow.UserID, conversationID)
if err != nil {
return err
}
// 1. 只有首次创建会话时才写首屏消息,避免同一 trigger 的重试把时间线重复刷一遍。
// 2. 若 timeline 已存在,说明这段主动调度会话已经被成功预热过,直接复用现成内容即可。
if baseSeq == 0 {
assistantText := resolveInitialActiveScheduleAssistantText(selectionResult, previewDetail)
if assistantText != "" {
if err := txAgentDAO.SaveChatHistoryInTx(ctx, triggerRow.UserID, conversationID, "assistant", assistantText, "", 0, 0); err != nil {
return err
}
if err := saveActiveScheduleTimelineEvent(ctx, txAgentDAO, triggerRow.UserID, conversationID, baseSeq+1, model.AgentTimelineKindAssistantText, "assistant", assistantText, nil); err != nil {
return err
}
baseSeq++
}
if shouldSeedActiveSchedulePreviewCard(selectionResult) {
cardPayload, err := buildActiveScheduleBusinessCardPayload(previewDetail)
if err != nil {
return err
}
if err := saveActiveScheduleTimelineEvent(ctx, txAgentDAO, triggerRow.UserID, conversationID, baseSeq+1, model.AgentTimelineKindBusinessCard, "assistant", assistantText, cardPayload); err != nil {
return err
}
}
}
sessionSnapshot := &model.ActiveScheduleSessionSnapshot{
SessionID: sessionID,
UserID: triggerRow.UserID,
ConversationID: conversationID,
TriggerID: triggerRow.ID,
CurrentPreviewID: strings.TrimSpace(previewDetail.PreviewID),
Status: resolveInitialActiveScheduleSessionStatus(selectionResult),
State: buildInitialActiveScheduleSessionState(selectionResult, previewDetail),
CreatedAt: now,
UpdatedAt: now,
}
return txSessionDAO.UpsertActiveScheduleSession(ctx, sessionSnapshot)
}
func buildActiveScheduleConversationID(triggerID string) string {
normalized := strings.TrimSpace(triggerID)
if normalized == "" {
return uuid.NewString()
}
return uuid.NewSHA1(activeScheduleConversationNamespace, []byte(normalized)).String()
}
func buildActiveScheduleSessionID(triggerID string) string {
normalized := strings.TrimSpace(triggerID)
if normalized == "" {
return uuid.NewString()
}
return uuid.NewSHA1(activeScheduleSessionNamespace, []byte(normalized)).String()
}
func ensureAgentConversationExists(ctx context.Context, agentDAO *dao.AgentDAO, userID int, conversationID string) error {
if agentDAO == nil {
return errors.New("agent dao 不能为空")
}
if userID <= 0 {
return fmt.Errorf("invalid user_id: %d", userID)
}
normalizedConversationID := strings.TrimSpace(conversationID)
if normalizedConversationID == "" {
return errors.New("conversation_id 不能为空")
}
exists, err := agentDAO.IfChatExists(ctx, userID, normalizedConversationID)
if err != nil {
return err
}
if exists {
return nil
}
_, err = agentDAO.CreateNewChat(userID, normalizedConversationID)
return err
}
func resolveInitialActiveScheduleAssistantText(selectionResult selection.Result, previewDetail activepreview.ActiveSchedulePreviewDetail) string {
switch selectionResult.Action {
case selection.ActionAskUser:
return firstNonEmptyString(
selectionResult.AskUserQuestion,
selectionResult.ExplanationText,
previewDetail.Explanation,
previewDetail.Notification,
"请先补充主动调度需要的关键信息。",
)
default:
return firstNonEmptyString(
selectionResult.ExplanationText,
selectionResult.NotificationSummary,
previewDetail.Notification,
previewDetail.Explanation,
"主动调度建议已更新。",
)
}
}
func shouldSeedActiveSchedulePreviewCard(selectionResult selection.Result) bool {
return selectionResult.Action == selection.ActionSelectCandidate
}
func resolveInitialActiveScheduleSessionStatus(selectionResult selection.Result) string {
switch selectionResult.Action {
case selection.ActionAskUser:
return model.ActiveScheduleSessionStatusWaitingUserReply
case selection.ActionSelectCandidate:
return model.ActiveScheduleSessionStatusReadyPreview
default:
return model.ActiveScheduleSessionStatusIgnored
}
}
func buildInitialActiveScheduleSessionState(
selectionResult selection.Result,
previewDetail activepreview.ActiveSchedulePreviewDetail,
) model.ActiveScheduleSessionState {
state := model.ActiveScheduleSessionState{
LastCandidateID: strings.TrimSpace(selectionResult.SelectedCandidateID),
MissingInfo: cloneStringSlice(previewDetail.ContextSummary.MissingInfo),
}
if !previewDetail.ExpiresAt.IsZero() {
expiresAt := previewDetail.ExpiresAt
state.ExpiresAt = &expiresAt
}
switch selectionResult.Action {
case selection.ActionAskUser:
state.PendingQuestion = firstNonEmptyString(
selectionResult.AskUserQuestion,
selectionResult.ExplanationText,
)
case selection.ActionSelectCandidate:
state.PendingQuestion = ""
state.MissingInfo = nil
state.FailedReason = ""
default:
state.PendingQuestion = ""
state.MissingInfo = nil
state.ExpiresAt = nil
}
return state
}
func buildActiveScheduleBusinessCardPayload(detail activepreview.ActiveSchedulePreviewDetail) (map[string]any, error) {
raw, err := json.Marshal(map[string]any{
"business_card": map[string]any{
"card_type": "active_schedule_preview",
"title": "SmartFlow 日程调整建议",
"summary": firstNonEmptyString(detail.Notification, detail.Explanation, detail.SelectedCandidate.Summary),
"data": detail,
},
})
if err != nil {
return nil, err
}
var payload map[string]any
if err := json.Unmarshal(raw, &payload); err != nil {
return nil, err
}
return payload, nil
}
func saveActiveScheduleTimelineEvent(
ctx context.Context,
agentDAO *dao.AgentDAO,
userID int,
conversationID string,
seq int64,
kind string,
role string,
content string,
payload map[string]any,
) error {
if agentDAO == nil {
return errors.New("agent dao 不能为空")
}
normalizedConversationID := strings.TrimSpace(conversationID)
if userID <= 0 || normalizedConversationID == "" {
return errors.New("时间线事件主键不合法")
}
payloadJSON := ""
if len(payload) > 0 {
raw, err := json.Marshal(payload)
if err != nil {
return err
}
payloadJSON = string(raw)
}
_, _, err := agentDAO.SaveConversationTimelineEvent(ctx, model.ChatTimelinePersistPayload{
UserID: userID,
ConversationID: normalizedConversationID,
Seq: seq,
Kind: kind,
Role: role,
Content: content,
PayloadJSON: payloadJSON,
TokensConsumed: 0,
})
return err
}
func firstNonEmptyString(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
func cloneStringSlice(values []string) []string {
if len(values) == 0 {
return nil
}
cloned := make([]string, len(values))
copy(cloned, values)
return cloned
}

View File

@@ -118,6 +118,7 @@ func BuildFeishuRequestedPayload(
requestedAt time.Time,
) sharedevents.FeishuNotificationRequestedPayload {
summary := strings.TrimSpace(notificationSummary)
targetURL := fmt.Sprintf("/assistant/%s", buildActiveScheduleConversationID(triggerRow.ID))
return sharedevents.FeishuNotificationRequestedPayload{
UserID: triggerRow.UserID,
TriggerID: triggerRow.ID,
@@ -126,9 +127,9 @@ func BuildFeishuRequestedPayload(
TargetType: triggerRow.TargetType,
TargetID: triggerRow.TargetID,
DedupeKey: BuildNotificationDedupeKey(triggerRow.UserID, triggerRow.TriggerType, triggerRow.RequestedAt),
TargetURL: fmt.Sprintf("/schedule-adjust/%s", strings.TrimSpace(previewID)),
TargetURL: targetURL,
SummaryText: summary,
FallbackText: buildNotificationFallbackText(summary, strings.TrimSpace(previewID)),
FallbackText: buildNotificationFallbackText(summary, targetURL),
TraceID: triggerRow.TraceID,
RequestedAt: requestedAt,
}
@@ -201,8 +202,8 @@ func normalizeKafkaConfig(cfg kafkabus.Config) kafkabus.Config {
return cfg
}
func buildNotificationFallbackText(summary string, previewID string) string {
link := fmt.Sprintf("/schedule-adjust/%s", previewID)
func buildNotificationFallbackText(summary string, targetURL string) string {
link := strings.TrimSpace(targetURL)
if summary == "" {
return "你有一条新的日程调整建议,请查看:" + link
}

View File

@@ -7,6 +7,7 @@ import (
"strings"
"time"
activegraph "github.com/LoveLosita/smartflow/backend/active_scheduler/graph"
activepreview "github.com/LoveLosita/smartflow/backend/active_scheduler/preview"
"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
"github.com/LoveLosita/smartflow/backend/dao"
@@ -28,38 +29,57 @@ const (
//
// 职责边界:
// 1. 只推进主动调度 trigger 的后台状态机,不负责启动 outbox worker
// 2. dry-run 与 preview 复用现有 service不再单独实现第二套候选生成逻辑;
// 2. dry-run 与选择器都复用 active_scheduler 独立模块,不再往 newAgent 里塞主动调度逻辑;
// 3. notification 只发布 requested 事件,不直接接真实飞书 provider。
type TriggerWorkflowService struct {
activeDAO *dao.ActiveScheduleDAO
dryRun *DryRunService
outbox *outboxinfra.Repository
kafkaCfg kafkabus.Config
clock func() time.Time
activeDAO *dao.ActiveScheduleDAO
graphRunner *activegraph.Runner
outbox *outboxinfra.Repository
kafkaCfg kafkabus.Config
agentDAO *dao.AgentDAO
sessionDAO *dao.ActiveScheduleSessionDAO
clock func() time.Time
}
func NewTriggerWorkflowService(
activeDAO *dao.ActiveScheduleDAO,
dryRun *DryRunService,
graphRunner *activegraph.Runner,
outboxRepo *outboxinfra.Repository,
kafkaCfg kafkabus.Config,
) (*TriggerWorkflowService, error) {
return NewTriggerWorkflowServiceWithOptions(activeDAO, graphRunner, outboxRepo, kafkaCfg)
}
// NewTriggerWorkflowServiceWithOptions 创建主动调度 trigger 编排服务,并允许注入迁移期可选能力。
func NewTriggerWorkflowServiceWithOptions(
activeDAO *dao.ActiveScheduleDAO,
graphRunner *activegraph.Runner,
outboxRepo *outboxinfra.Repository,
kafkaCfg kafkabus.Config,
opts ...TriggerWorkflowOption,
) (*TriggerWorkflowService, error) {
if activeDAO == nil {
return nil, errors.New("active schedule dao 不能为空")
}
if dryRun == nil {
return nil, errors.New("dry-run service 不能为空")
if graphRunner == nil {
return nil, errors.New("active scheduler graph runner 不能为空")
}
if outboxRepo == nil {
return nil, errors.New("outbox repository 不能为空")
}
return &TriggerWorkflowService{
activeDAO: activeDAO,
dryRun: dryRun,
outbox: outboxRepo,
kafkaCfg: kafkaCfg,
clock: time.Now,
}, nil
svc := &TriggerWorkflowService{
activeDAO: activeDAO,
graphRunner: graphRunner,
outbox: outboxRepo,
kafkaCfg: kafkaCfg,
clock: time.Now,
}
for _, opt := range opts {
if opt != nil {
opt(svc)
}
}
return svc, nil
}
func (s *TriggerWorkflowService) SetClock(clock func() time.Time) {
@@ -68,12 +88,15 @@ func (s *TriggerWorkflowService) SetClock(clock func() time.Time) {
}
}
// TriggerWorkflowOption 是 trigger 编排服务的可选注入项。
type TriggerWorkflowOption func(*TriggerWorkflowService)
// ProcessTriggeredInTx 在 outbox 消费事务内推进 trigger 主链路。
//
// 步骤化说明:
// 1. 先锁 trigger 行,确保同一 trigger 在并发 worker 下只能由一个事务推进;
// 2. 再把状态切到 processing避免排障时看不出消息已经被消费
// 3. 复用 dry-run + preview service 生成预览;若发现已有 preview则直接复用避免重复写库
// 3. 复用 active scheduler graph 跑 dry-run + 受限选择;若发现已有 preview则直接复用避免重复写库
// 4. preview 成功后回写 trigger 状态,并在同一事务里补发 notification.requested outbox
// 5. 任一步失败都返回 error由外层 handler 负责记录 failed 状态并触发 outbox retry。
func (s *TriggerWorkflowService) ProcessTriggeredInTx(
@@ -81,7 +104,7 @@ func (s *TriggerWorkflowService) ProcessTriggeredInTx(
tx *gorm.DB,
payload sharedevents.ActiveScheduleTriggeredPayload,
) error {
if s == nil || s.activeDAO == nil || s.dryRun == nil || s.outbox == nil {
if s == nil || s.activeDAO == nil || s.graphRunner == nil || s.outbox == nil {
return errors.New("trigger workflow service 未初始化")
}
if tx == nil {
@@ -125,14 +148,18 @@ func (s *TriggerWorkflowService) ProcessTriggeredInTx(
}
domainTrigger := buildDomainTriggerFromModel(*triggerRow, payload)
dryRunResult, err := s.dryRun.DryRun(ctx, domainTrigger)
graphResult, err := s.graphRunner.Run(ctx, domainTrigger)
if err != nil {
return err
}
if len(dryRunResult.Candidates) == 0 {
if graphResult == nil || graphResult.DryRunData == nil {
return errors.New("active scheduler graph 返回空结果")
}
dryRunData := graphResult.DryRunData
if len(dryRunData.Candidates) == 0 {
return s.markClosedWithoutPreview(ctx, txDAO, triggerRow.ID, now)
}
if !dryRunResult.Observation.Decision.ShouldNotify && !dryRunResult.Observation.Decision.ShouldWritePreview {
if !dryRunData.Observation.Decision.ShouldNotify && !dryRunData.Observation.Decision.ShouldWritePreview {
return s.markClosedWithoutPreview(ctx, txDAO, triggerRow.ID, now)
}
@@ -141,11 +168,15 @@ func (s *TriggerWorkflowService) ProcessTriggeredInTx(
return err
}
previewResp, err := previewService.CreatePreview(ctx, activepreview.CreatePreviewRequest{
ActiveContext: dryRunResult.Context,
Observation: dryRunResult.Observation,
Candidates: dryRunResult.Candidates,
TriggerID: triggerRow.ID,
GeneratedAt: now,
ActiveContext: dryRunData.Context,
Observation: dryRunData.Observation,
Candidates: dryRunData.Candidates,
TriggerID: triggerRow.ID,
GeneratedAt: now,
SelectedCandidateID: graphResult.SelectionResult.SelectedCandidateID,
ExplanationText: graphResult.SelectionResult.ExplanationText,
NotificationSummary: graphResult.SelectionResult.NotificationSummary,
FallbackUsed: graphResult.SelectionResult.FallbackUsed,
})
if err != nil {
return err
@@ -162,10 +193,16 @@ func (s *TriggerWorkflowService) ProcessTriggeredInTx(
return err
}
if !dryRunResult.Observation.Decision.ShouldNotify {
if !dryRunData.Observation.Decision.ShouldNotify {
return nil
}
// 1. 离线通知发出前,先把用户点击后要进入的助手会话和主动调度 session 预热好。
// 2. 这一步和 preview / notification outbox 在同一事务内提交,避免出现“飞书已送达但会话空白”的断裂状态。
if err := s.bootstrapActiveScheduleConversationInTx(ctx, tx, *triggerRow, previewResp.Detail, graphResult.SelectionResult, now); err != nil {
return err
}
notificationPayload := BuildFeishuRequestedPayload(
*triggerRow,
previewID,

View File

@@ -156,7 +156,7 @@ func (api *AgentHandler) ChatAgent(c *gin.Context) {
// 设计说明:
// 1) 该接口用于配合 SSE 聊天链路:标题异步生成后,前端可通过 conversation_id 拉取;
// 2) 不依赖 SSE header 动态更新避免“header 必须首包前写入”的协议限制;
// 3) 会话不存在时返回 400,避免前端把无效会话当成系统错误。
// 3) 会话不存在或不属于当前用户时返回 404,避免前端把无效会话误判成参数类型错误。
func (api *AgentHandler) GetConversationMeta(c *gin.Context) {
// 1. 读取 query 参数并做基础校验。
conversationID := strings.TrimSpace(c.Query("conversation_id"))
@@ -175,9 +175,9 @@ func (api *AgentHandler) GetConversationMeta(c *gin.Context) {
// 4. 调 service 查询会话元信息。
meta, err := api.svc.GetConversationMeta(ctx, userID, conversationID)
if err != nil {
// 会话不存在按参数错误处理,返回 400 给前端更直观
// 会话不存在或越权访问时返回 404让前端能和“参数格式错误”区分开
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusBadRequest, respond.WrongParamType)
c.JSON(http.StatusNotFound, respond.ConversationNotFound)
return
}
respond.DealWithError(c, err)
@@ -255,7 +255,7 @@ func (api *AgentHandler) GetConversationList(c *gin.Context) {
// 说明:
// 1. 该接口是新前端刷新重建的单一来源;
// 2. 返回结果已按 seq 升序,前端按数组顺序渲染即可;
// 3. 会话不存在时统一返回 400,避免误判成系统异常
// 3. 会话不存在或不属于当前用户时统一返回 404,避免误判成参数格式问题
func (api *AgentHandler) GetConversationTimeline(c *gin.Context) {
conversationID := strings.TrimSpace(c.Query("conversation_id"))
if conversationID == "" {
@@ -271,7 +271,7 @@ func (api *AgentHandler) GetConversationTimeline(c *gin.Context) {
timeline, err := api.svc.GetConversationTimeline(ctx, userID, conversationID)
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusBadRequest, respond.WrongParamType)
c.JSON(http.StatusNotFound, respond.ConversationNotFound)
return
}
respond.DealWithError(c, err)

View File

@@ -2,6 +2,7 @@ package cmd
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
@@ -12,9 +13,12 @@ import (
activeadapters "github.com/LoveLosita/smartflow/backend/active_scheduler/adapters"
"github.com/LoveLosita/smartflow/backend/active_scheduler/applyadapter"
activegraph "github.com/LoveLosita/smartflow/backend/active_scheduler/graph"
activejob "github.com/LoveLosita/smartflow/backend/active_scheduler/job"
activepreview "github.com/LoveLosita/smartflow/backend/active_scheduler/preview"
activesel "github.com/LoveLosita/smartflow/backend/active_scheduler/selection"
activesvc "github.com/LoveLosita/smartflow/backend/active_scheduler/service"
activeTrigger "github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
"github.com/LoveLosita/smartflow/backend/api"
"github.com/LoveLosita/smartflow/backend/dao"
kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
@@ -30,12 +34,14 @@ import (
"github.com/LoveLosita/smartflow/backend/model"
newagentconv "github.com/LoveLosita/smartflow/backend/newAgent/conv"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
newagentstream "github.com/LoveLosita/smartflow/backend/newAgent/stream"
newagenttools "github.com/LoveLosita/smartflow/backend/newAgent/tools"
"github.com/LoveLosita/smartflow/backend/newAgent/tools/web"
"github.com/LoveLosita/smartflow/backend/notification"
"github.com/LoveLosita/smartflow/backend/pkg"
"github.com/LoveLosita/smartflow/backend/routers"
"github.com/LoveLosita/smartflow/backend/service"
agentsvcsvc "github.com/LoveLosita/smartflow/backend/service/agentsvc"
eventsvc "github.com/LoveLosita/smartflow/backend/service/events"
"github.com/go-redis/redis/v8"
"github.com/spf13/viper"
@@ -211,7 +217,17 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) {
courseService := buildCourseService(courseRepo, scheduleRepo)
taskClassService := service.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager)
scheduleService := service.NewScheduleService(scheduleRepo, userRepo, taskClassRepo, manager, cacheRepo)
agentService := service.NewAgentServiceWithSchedule(aiHub, agentRepo, taskRepo, cacheRepo, agentCacheRepo, eventBus, scheduleService, taskSv)
agentService := service.NewAgentServiceWithSchedule(
aiHub,
agentRepo,
taskRepo,
cacheRepo,
agentCacheRepo,
manager.ActiveScheduleSession,
eventBus,
scheduleService,
taskSv,
)
configureAgentService(
agentService,
@@ -238,6 +254,15 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) {
if err != nil {
return nil, err
}
// 1. 主动调度选择器单独复用 Pro 模型LLM 失败时由 selection 层显式回退到确定性候选;
// 2. dry-run 与 selection 通过 graph runner 串起来,避免 trigger_pipeline 再拼第二套候选逻辑。
activeScheduleLLMClient := infrallm.WrapArkClient(aiHub.Pro)
activeScheduleSelector := activesel.NewService(activeScheduleLLMClient)
activeScheduleGraphRunner, err := activegraph.NewRunner(activeScheduleDryRun.AsGraphDryRunFunc(), activeScheduleSelector)
if err != nil {
return nil, err
}
agentService.SetActiveScheduleSessionRerunFunc(buildActiveScheduleSessionRerunFunc(manager.ActiveSchedule, activeScheduleGraphRunner, activeSchedulePreviewConfirm))
// 1. 生产投递先切到用户级飞书 Webhook providermock provider 文件继续保留给后续单测和本地隔离验证。
// 2. provider 与配置测试接口共用同一个实例,保证“测试成功”和“正式投递”走同一套 URL 校验、JSON 拼装和 HTTP 结果分类。
feishuProvider, err := notification.NewWebhookFeishuProvider(manager.Notification, notification.WebhookFeishuProviderOptions{
@@ -257,7 +282,13 @@ func buildRuntime(ctx context.Context) (*appRuntime, error) {
var activeTriggerWorkflow *activesvc.TriggerWorkflowService
var activeJobScanner *activejob.Scanner
if eventBus != nil {
activeTriggerWorkflow, err = activesvc.NewTriggerWorkflowService(manager.ActiveSchedule, activeScheduleDryRun, outboxRepo, kafkabus.LoadConfig())
activeTriggerWorkflow, err = activesvc.NewTriggerWorkflowServiceWithOptions(
manager.ActiveSchedule,
activeScheduleGraphRunner,
outboxRepo,
kafkabus.LoadConfig(),
activesvc.WithActiveScheduleSessionBridge(manager.Agent, manager.ActiveScheduleSession),
)
if err != nil {
return nil, err
}
@@ -359,6 +390,166 @@ func buildActiveSchedulePreviewConfirmService(db *gorm.DB, activeDAO *dao.Active
return activesvc.NewPreviewConfirmService(dryRun, previewService, activeDAO, applyadapter.NewGormApplyAdapter(db))
}
// buildActiveScheduleSessionRerunFunc 把主动调度 graph / preview 能力装成聊天入口可调用的 rerun 闭包。
//
// 说明:
// 1. 这里只做最小接线:复用现有 trigger -> graph -> preview 组件,不把 worker/notification 再搬一遍;
// 2. 成功时返回 session 状态、assistant 文本和业务卡片数据;
// 3. 失败时直接把 error 交回聊天入口,由上层统一写失败日志和 SSE 错误。
func buildActiveScheduleSessionRerunFunc(
activeDAO *dao.ActiveScheduleDAO,
graphRunner *activegraph.Runner,
previewConfirm *activesvc.PreviewConfirmService,
) agentsvcsvc.ActiveScheduleSessionRerunFunc {
return func(
ctx context.Context,
session *model.ActiveScheduleSessionSnapshot,
userMessage string,
traceID string,
requestStart time.Time,
) (*agentsvcsvc.ActiveScheduleSessionRerunResult, error) {
if activeDAO == nil || graphRunner == nil || previewConfirm == nil {
return nil, fmt.Errorf("主动调度 rerun 依赖未初始化")
}
if session == nil {
return nil, fmt.Errorf("主动调度 session 不能为空")
}
triggerRow, err := activeDAO.GetTriggerByID(ctx, session.TriggerID)
if err != nil {
return nil, err
}
// 1. 当前最小接线先复用“原 trigger + 最新数据库事实”重跑 active scheduler graph。
// 2. 用户这次回复的正文已经由聊天入口写进 conversation/timeline但还没有下沉到 active_scheduler readers。
// 3. 后续若要让 ask_user 回复直接改写 graph 事实源,应在 reader/context builder 层继续补这一跳。
domainTrigger := activeTrigger.ActiveScheduleTrigger{
TriggerID: triggerRow.ID,
UserID: triggerRow.UserID,
TriggerType: activeTrigger.TriggerType(triggerRow.TriggerType),
Source: activeTrigger.SourceUserFeedback,
TargetType: activeTrigger.TargetType(triggerRow.TargetType),
TargetID: triggerRow.TargetID,
FeedbackID: triggerRow.FeedbackID,
IdempotencyKey: triggerRow.IdempotencyKey,
MockNow: nil,
IsMockTime: false,
RequestedAt: requestStart,
TraceID: traceID,
}
if err := domainTrigger.Validate(); err != nil {
return nil, err
}
graphResult, err := graphRunner.Run(ctx, domainTrigger)
if err != nil {
return nil, err
}
if graphResult == nil || graphResult.DryRunData == nil || graphResult.DryRunData.Context == nil {
return nil, fmt.Errorf("主动调度 graph 返回空结果")
}
selectionResult := graphResult.SelectionResult
state := session.State
state.LastCandidateID = strings.TrimSpace(selectionResult.SelectedCandidateID)
state.LastNotificationID = ""
state.FailedReason = ""
state.MissingInfo = cloneStringSlice(graphResult.DryRunData.Context.DerivedFacts.MissingInfo)
switch selectionResult.Action {
case activesel.ActionSelectCandidate:
if !graphResult.DryRunData.Observation.Decision.ShouldWritePreview {
return nil, fmt.Errorf("主动调度 graph 选择了候选,但未产出可写 preview")
}
previewResp, err := previewConfirm.CreatePreviewFromDryRun(ctx, activepreview.CreatePreviewRequest{
ActiveContext: graphResult.DryRunData.Context,
Observation: graphResult.DryRunData.Observation,
Candidates: graphResult.DryRunData.Candidates,
TriggerID: triggerRow.ID,
GeneratedAt: requestStart,
SelectedCandidateID: selectionResult.SelectedCandidateID,
ExplanationText: selectionResult.ExplanationText,
NotificationSummary: selectionResult.NotificationSummary,
FallbackUsed: selectionResult.FallbackUsed,
})
if err != nil {
return nil, err
}
state.PendingQuestion = ""
state.MissingInfo = nil
state.FailedReason = ""
expiresAt := previewResp.Detail.ExpiresAt
state.ExpiresAt = &expiresAt
return &agentsvcsvc.ActiveScheduleSessionRerunResult{
AssistantText: firstNonEmptyString(selectionResult.ExplanationText, selectionResult.NotificationSummary, previewResp.Detail.Explanation, previewResp.Detail.Notification, "主动调度建议已更新。"),
BusinessCard: &newagentstream.StreamBusinessCardExtra{
CardType: "active_schedule_preview",
Title: "SmartFlow 日程调整建议",
Summary: firstNonEmptyString(selectionResult.NotificationSummary, previewResp.Detail.Notification, previewResp.Detail.Explanation),
Data: previewDetailToMap(previewResp.Detail),
},
SessionState: state,
SessionStatus: model.ActiveScheduleSessionStatusReadyPreview,
PreviewID: previewResp.Detail.PreviewID,
}, nil
case activesel.ActionAskUser:
question := firstNonEmptyString(selectionResult.AskUserQuestion, selectionResult.ExplanationText, "请继续补充主动调度需要的信息。")
state.PendingQuestion = question
state.ExpiresAt = nil
return &agentsvcsvc.ActiveScheduleSessionRerunResult{
AssistantText: question,
SessionState: state,
SessionStatus: model.ActiveScheduleSessionStatusWaitingUserReply,
}, nil
default:
assistantText := firstNonEmptyString(selectionResult.ExplanationText, selectionResult.NotificationSummary, "当前主动调度暂时没有需要继续处理的内容。")
state.PendingQuestion = ""
state.MissingInfo = nil
state.ExpiresAt = nil
return &agentsvcsvc.ActiveScheduleSessionRerunResult{
AssistantText: assistantText,
SessionState: state,
SessionStatus: model.ActiveScheduleSessionStatusIgnored,
}, nil
}
}
}
// previewDetailToMap 将 active_schedule preview 详情转成通用 map供 timeline business_card 直接复用。
func previewDetailToMap(detail activepreview.ActiveSchedulePreviewDetail) map[string]any {
raw, err := json.Marshal(detail)
if err != nil {
return map[string]any{}
}
var output map[string]any
if err := json.Unmarshal(raw, &output); err != nil {
return map[string]any{}
}
return output
}
// firstNonEmptyString 负责在一组候选文本里挑出第一条可展示内容。
func firstNonEmptyString(values ...string) string {
for _, value := range values {
if trimmed := strings.TrimSpace(value); trimmed != "" {
return trimmed
}
}
return ""
}
// cloneStringSlice 负责复制 string 切片,避免直接复用底层数组被后续修改。
func cloneStringSlice(values []string) []string {
if len(values) == 0 {
return nil
}
copied := make([]string, len(values))
copy(copied, values)
return copied
}
func configureAgentService(
agentService *service.AgentService,
ragRuntime infrarag.Runtime,
@@ -487,12 +678,13 @@ func buildTaskClassUpsertFunc(taskClassRepo *dao.TaskClassDAO) func(userID int,
}
}
func buildQuickTaskCreateFunc(taskRepo *dao.TaskDAO) func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) {
return func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) {
func buildQuickTaskCreateFunc(taskRepo *dao.TaskDAO) func(userID int, title string, priorityGroup int, estimatedSections int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) {
return func(userID int, title string, priorityGroup int, estimatedSections int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (int, error) {
created, err := taskRepo.AddTask(&model.Task{
UserID: userID,
Title: title,
Priority: priorityGroup,
EstimatedSections: model.NormalizeEstimatedSections(&estimatedSections),
IsCompleted: false,
DeadlineAt: deadlineAt,
UrgencyThresholdAt: urgencyThresholdAt,
@@ -528,11 +720,12 @@ func buildQuickTaskQueryFunc(agentService *service.AgentService) func(ctx contex
deadlineStr = r.DeadlineAt.In(time.Local).Format("2006-01-02 15:04")
}
results = append(results, newagentmodel.TaskQueryResult{
ID: r.ID,
Title: r.Title,
PriorityGroup: r.PriorityGroup,
IsCompleted: r.IsCompleted,
DeadlineAt: deadlineStr,
ID: r.ID,
Title: r.Title,
PriorityGroup: r.PriorityGroup,
EstimatedSections: model.NormalizeEstimatedSections(&r.EstimatedSections),
IsCompleted: r.IsCompleted,
DeadlineAt: deadlineStr,
})
}
return results, nil

View File

@@ -8,10 +8,11 @@ import (
func UserAddTaskRequestToModel(request *model.UserAddTaskRequest, userID int) *model.Task {
return &model.Task{
Title: request.Title,
Priority: request.PriorityGroup,
DeadlineAt: request.DeadlineAt,
UserID: userID,
Title: request.Title,
Priority: request.PriorityGroup,
EstimatedSections: model.NormalizeEstimatedSections(&request.EstimatedSections),
DeadlineAt: request.DeadlineAt,
UserID: userID,
}
}
@@ -21,12 +22,13 @@ func ModelToUserAddTaskResponse(task *model.Task) *model.UserAddTaskResponse {
status = "completed"
}
return &model.UserAddTaskResponse{
ID: task.ID,
Title: task.Title,
PriorityGroup: task.Priority,
DeadlineAt: task.DeadlineAt,
Status: status,
CreatedAt: time.Now(), // 创建时间为当前时间
ID: task.ID,
Title: task.Title,
PriorityGroup: task.Priority,
EstimatedSections: model.NormalizeEstimatedSections(&task.EstimatedSections),
DeadlineAt: task.DeadlineAt,
Status: status,
CreatedAt: time.Now(), // 创建时间为当前时间
}
}
@@ -53,6 +55,7 @@ func ModelToGetUserTasksResp(tasks []model.Task) []model.GetUserTaskResp {
UserID: task.UserID,
Title: task.Title,
PriorityGroup: task.Priority,
EstimatedSections: model.NormalizeEstimatedSections(&task.EstimatedSections),
Status: status,
Deadline: deadline,
IsCompleted: task.IsCompleted,
@@ -81,6 +84,7 @@ func ModelToGetUserTaskResp(task *model.Task) model.GetUserTaskResp {
UserID: task.UserID,
Title: task.Title,
PriorityGroup: task.Priority,
EstimatedSections: model.NormalizeEstimatedSections(&task.EstimatedSections),
Status: status,
Deadline: deadline,
IsCompleted: task.IsCompleted,

View File

@@ -0,0 +1,400 @@
package dao
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/model"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
var activeScheduleSessionLiveStatuses = []string{
model.ActiveScheduleSessionStatusWaitingUserReply,
model.ActiveScheduleSessionStatusRerunning,
}
// ActiveScheduleSessionDAO 负责主动调度会话的数据库读写。
//
// 职责边界:
// 1. 只管 session 表本身,不管聊天入口拦截策略;
// 2. 只提供按 session_id / conversation_id 的读写能力,不编排 graph
// 3. cache 命中策略由上层决定,这里始终把 MySQL 当作最终真相。
type ActiveScheduleSessionDAO struct {
db *gorm.DB
}
// NewActiveScheduleSessionDAO 创建主动调度会话 DAO。
func NewActiveScheduleSessionDAO(db *gorm.DB) *ActiveScheduleSessionDAO {
return &ActiveScheduleSessionDAO{db: db}
}
// WithTx 基于外部事务句柄构造同事务 DAO。
func (d *ActiveScheduleSessionDAO) WithTx(tx *gorm.DB) *ActiveScheduleSessionDAO {
return &ActiveScheduleSessionDAO{db: tx}
}
func (d *ActiveScheduleSessionDAO) ensureDB() error {
if d == nil || d.db == nil {
return errors.New("active schedule session dao 未初始化")
}
return nil
}
// UpsertActiveScheduleSession 按 session_id 幂等写入或覆盖主动调度会话。
//
// 步骤化说明:
// 1. 先校验主键、归属用户和状态,避免把脏会话写进数据表;
// 2. 再把轻量 state 统一序列化为 state_json保证数据库侧格式稳定
// 3. 最后走 OnConflict upsert保留 created_at仅刷新业务字段和 updated_at。
func (d *ActiveScheduleSessionDAO) UpsertActiveScheduleSession(ctx context.Context, snapshot *model.ActiveScheduleSessionSnapshot) error {
if err := d.ensureDB(); err != nil {
return err
}
normalized, err := normalizeActiveScheduleSessionSnapshot(snapshot)
if err != nil {
return err
}
stateJSON, err := marshalActiveScheduleSessionState(normalized.State)
if err != nil {
return fmt.Errorf("marshal active schedule session state failed: %w", err)
}
now := time.Now()
row := model.ActiveScheduleSession{
SessionID: normalized.SessionID,
UserID: normalized.UserID,
ConversationID: nullableStringPtr(normalized.ConversationID),
TriggerID: normalized.TriggerID,
CurrentPreviewID: nullableStringPtr(normalized.CurrentPreviewID),
Status: normalized.Status,
StateJSON: stateJSON,
CreatedAt: normalized.CreatedAt,
UpdatedAt: now,
}
if row.CreatedAt.IsZero() {
row.CreatedAt = now
}
return d.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{
{Name: "session_id"},
},
DoUpdates: clause.Assignments(map[string]any{
"user_id": row.UserID,
"conversation_id": row.ConversationID,
"trigger_id": row.TriggerID,
"current_preview_id": row.CurrentPreviewID,
"status": row.Status,
"state_json": row.StateJSON,
"updated_at": row.UpdatedAt,
}),
}).Create(&row).Error
}
// GetActiveScheduleSessionBySessionID 按 session_id 读取任意状态的会话记录。
//
// 返回语义:
// 1. 命中:返回 snapshot, nil
// 2. 未命中:返回 nil, nil交给上层判断是否需要走回源或新建
// 3. 数据损坏:返回 error避免把坏状态继续传给拦截逻辑。
func (d *ActiveScheduleSessionDAO) GetActiveScheduleSessionBySessionID(ctx context.Context, sessionID string) (*model.ActiveScheduleSessionSnapshot, error) {
if err := d.ensureDB(); err != nil {
return nil, err
}
normalizedSessionID := strings.TrimSpace(sessionID)
if normalizedSessionID == "" {
return nil, errors.New("session_id is empty")
}
var row model.ActiveScheduleSession
err := d.db.WithContext(ctx).
Where("session_id = ?", normalizedSessionID).
First(&row).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, err
}
return activeScheduleSessionSnapshotFromRow(&row)
}
// GetActiveScheduleSessionByConversationID 按 user_id + conversation_id 读取最新的会话记录。
//
// 职责边界:
// 1. 始终返回同一 conversation 最新的一条记录,方便上层直接判断当前 status
// 2. 不在 DAO 内部做“是否拦截”的业务裁决,避免把路由规则写死在存储层;
// 3. 若同一 conversation 误写出多条记录,按最近更新时间优先返回。
func (d *ActiveScheduleSessionDAO) GetActiveScheduleSessionByConversationID(ctx context.Context, userID int, conversationID string) (*model.ActiveScheduleSessionSnapshot, error) {
if err := d.ensureDB(); err != nil {
return nil, err
}
if userID <= 0 {
return nil, fmt.Errorf("invalid user_id: %d", userID)
}
normalizedConversationID := strings.TrimSpace(conversationID)
if normalizedConversationID == "" {
return nil, errors.New("conversation_id is empty")
}
var row model.ActiveScheduleSession
err := d.db.WithContext(ctx).
Where("user_id = ? AND conversation_id = ?", userID, normalizedConversationID).
Order("updated_at DESC, created_at DESC, session_id DESC").
First(&row).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, nil
}
return nil, err
}
return activeScheduleSessionSnapshotFromRow(&row)
}
// UpdateActiveScheduleSessionFieldsBySessionID 按 session_id 更新局部字段。
//
// 说明:
// 1. 这里不负责 state_json 的序列化,调用方需要自己准备好最终字段值;
// 2. 若 updates 为空,直接返回 nil避免多余的数据库写入
// 3. updated_at 会在这里自动刷新,保证时间线可追踪。
func (d *ActiveScheduleSessionDAO) UpdateActiveScheduleSessionFieldsBySessionID(ctx context.Context, sessionID string, updates map[string]any) error {
if err := d.ensureDB(); err != nil {
return err
}
normalizedSessionID := strings.TrimSpace(sessionID)
if normalizedSessionID == "" {
return errors.New("session_id is empty")
}
if len(updates) == 0 {
return nil
}
normalizedUpdates := cloneUpdateMap(updates)
if _, ok := normalizedUpdates["updated_at"]; !ok {
normalizedUpdates["updated_at"] = time.Now()
}
return d.db.WithContext(ctx).
Model(&model.ActiveScheduleSession{}).
Where("session_id = ?", normalizedSessionID).
Updates(normalizedUpdates).Error
}
// UpdateActiveScheduleSessionFieldsByConversationID 按 user_id + conversation_id 更新最新记录的局部字段。
//
// 步骤化说明:
// 1. 先定位同一 conversation 最新的 session再按 session_id 回写,避免一次 update 覆盖多条历史;
// 2. 再写入局部字段和 updated_at保证状态变化可以按会话维度回写
// 3. 找不到任何会话时直接返回,交给上层决定是否要新建 session 或释放普通聊天。
func (d *ActiveScheduleSessionDAO) UpdateActiveScheduleSessionFieldsByConversationID(ctx context.Context, userID int, conversationID string, updates map[string]any) error {
if err := d.ensureDB(); err != nil {
return err
}
if userID <= 0 {
return fmt.Errorf("invalid user_id: %d", userID)
}
normalizedConversationID := strings.TrimSpace(conversationID)
if normalizedConversationID == "" {
return errors.New("conversation_id is empty")
}
if len(updates) == 0 {
return nil
}
row, err := d.GetActiveScheduleSessionByConversationID(ctx, userID, normalizedConversationID)
if err != nil {
return err
}
if row == nil {
return gorm.ErrRecordNotFound
}
normalizedUpdates := cloneUpdateMap(updates)
if _, ok := normalizedUpdates["updated_at"]; !ok {
normalizedUpdates["updated_at"] = time.Now()
}
return d.db.WithContext(ctx).
Model(&model.ActiveScheduleSession{}).
Where("session_id = ?", row.SessionID).
Updates(normalizedUpdates).Error
}
func normalizeActiveScheduleSessionSnapshot(snapshot *model.ActiveScheduleSessionSnapshot) (*model.ActiveScheduleSessionSnapshot, error) {
if snapshot == nil {
return nil, errors.New("active schedule session snapshot is nil")
}
normalizedSessionID := strings.TrimSpace(snapshot.SessionID)
if normalizedSessionID == "" {
return nil, errors.New("session_id is empty")
}
if snapshot.UserID <= 0 {
return nil, fmt.Errorf("invalid user_id: %d", snapshot.UserID)
}
normalizedStatus, err := normalizeActiveScheduleSessionStatus(snapshot.Status)
if err != nil {
return nil, err
}
normalizedTriggerID := strings.TrimSpace(snapshot.TriggerID)
if normalizedTriggerID == "" {
return nil, errors.New("trigger_id is empty")
}
normalized := *snapshot
normalized.SessionID = normalizedSessionID
normalized.UserID = snapshot.UserID
normalized.ConversationID = strings.TrimSpace(snapshot.ConversationID)
normalized.TriggerID = normalizedTriggerID
normalized.CurrentPreviewID = strings.TrimSpace(snapshot.CurrentPreviewID)
normalized.Status = normalizedStatus
normalized.State = normalizeActiveScheduleSessionState(snapshot.State)
return &normalized, nil
}
func normalizeActiveScheduleSessionStatus(raw string) (string, error) {
switch strings.ToLower(strings.TrimSpace(raw)) {
case model.ActiveScheduleSessionStatusWaitingUserReply:
return model.ActiveScheduleSessionStatusWaitingUserReply, nil
case model.ActiveScheduleSessionStatusRerunning:
return model.ActiveScheduleSessionStatusRerunning, nil
case model.ActiveScheduleSessionStatusReadyPreview:
return model.ActiveScheduleSessionStatusReadyPreview, nil
case model.ActiveScheduleSessionStatusApplied:
return model.ActiveScheduleSessionStatusApplied, nil
case model.ActiveScheduleSessionStatusIgnored:
return model.ActiveScheduleSessionStatusIgnored, nil
case model.ActiveScheduleSessionStatusExpired:
return model.ActiveScheduleSessionStatusExpired, nil
case model.ActiveScheduleSessionStatusFailed:
return model.ActiveScheduleSessionStatusFailed, nil
default:
return "", fmt.Errorf("invalid active schedule session status: %s", raw)
}
}
func normalizeActiveScheduleSessionState(state model.ActiveScheduleSessionState) model.ActiveScheduleSessionState {
state.PendingQuestion = strings.TrimSpace(state.PendingQuestion)
state.LastCandidateID = strings.TrimSpace(state.LastCandidateID)
state.LastNotificationID = strings.TrimSpace(state.LastNotificationID)
state.FailedReason = strings.TrimSpace(state.FailedReason)
if state.ExpiresAt != nil && state.ExpiresAt.IsZero() {
state.ExpiresAt = nil
}
if len(state.MissingInfo) > 0 {
state.MissingInfo = dedupeAndTrimStrings(state.MissingInfo)
}
return state
}
func marshalActiveScheduleSessionState(state model.ActiveScheduleSessionState) (string, error) {
normalized := normalizeActiveScheduleSessionState(state)
raw, err := json.Marshal(normalized)
if err != nil {
return "", err
}
text := strings.TrimSpace(string(raw))
if text == "" {
return "{}", nil
}
return text, nil
}
func unmarshalActiveScheduleSessionState(raw string) (model.ActiveScheduleSessionState, error) {
clean := strings.TrimSpace(raw)
if clean == "" || clean == "null" {
return model.ActiveScheduleSessionState{}, nil
}
var state model.ActiveScheduleSessionState
if err := json.Unmarshal([]byte(clean), &state); err != nil {
return model.ActiveScheduleSessionState{}, err
}
state = normalizeActiveScheduleSessionState(state)
return state, nil
}
func activeScheduleSessionSnapshotFromRow(row *model.ActiveScheduleSession) (*model.ActiveScheduleSessionSnapshot, error) {
if row == nil {
return nil, errors.New("active schedule session row is nil")
}
state, err := unmarshalActiveScheduleSessionState(row.StateJSON)
if err != nil {
return nil, fmt.Errorf("unmarshal active schedule session state failed: %w", err)
}
return &model.ActiveScheduleSessionSnapshot{
SessionID: row.SessionID,
UserID: row.UserID,
ConversationID: nullableStringValue(row.ConversationID),
TriggerID: row.TriggerID,
CurrentPreviewID: nullableStringValue(row.CurrentPreviewID),
Status: row.Status,
State: state,
CreatedAt: row.CreatedAt,
UpdatedAt: row.UpdatedAt,
}, nil
}
func nullableStringPtr(raw string) *string {
normalized := strings.TrimSpace(raw)
if normalized == "" {
return nil
}
return &normalized
}
func nullableStringValue(raw *string) string {
if raw == nil {
return ""
}
return strings.TrimSpace(*raw)
}
func cloneUpdateMap(updates map[string]any) map[string]any {
cloned := make(map[string]any, len(updates)+1)
for key, value := range updates {
cloned[key] = value
}
return cloned
}
func dedupeAndTrimStrings(values []string) []string {
if len(values) == 0 {
return nil
}
result := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, item := range values {
normalized := strings.TrimSpace(item)
if normalized == "" {
continue
}
if _, ok := seen[normalized]; ok {
continue
}
seen[normalized] = struct{}{}
result = append(result, normalized)
}
if len(result) == 0 {
return nil
}
return result
}

View File

@@ -8,28 +8,30 @@ import (
// RepoManager 聚合所有 DAO供服务层做跨仓储事务编排。
type RepoManager struct {
db *gorm.DB
Schedule *ScheduleDAO
Task *TaskDAO
Course *CourseDAO
TaskClass *TaskClassDAO
User *UserDAO
Agent *AgentDAO
ActiveSchedule *ActiveScheduleDAO
Notification *NotificationChannelDAO
db *gorm.DB
Schedule *ScheduleDAO
Task *TaskDAO
Course *CourseDAO
TaskClass *TaskClassDAO
User *UserDAO
Agent *AgentDAO
ActiveSchedule *ActiveScheduleDAO
ActiveScheduleSession *ActiveScheduleSessionDAO
Notification *NotificationChannelDAO
}
func NewManager(db *gorm.DB) *RepoManager {
return &RepoManager{
db: db,
Schedule: NewScheduleDAO(db),
Task: NewTaskDAO(db),
Course: NewCourseDAO(db),
TaskClass: NewTaskClassDAO(db),
User: NewUserDAO(db),
Agent: NewAgentDAO(db),
ActiveSchedule: NewActiveScheduleDAO(db),
Notification: NewNotificationChannelDAO(db),
db: db,
Schedule: NewScheduleDAO(db),
Task: NewTaskDAO(db),
Course: NewCourseDAO(db),
TaskClass: NewTaskClassDAO(db),
User: NewUserDAO(db),
Agent: NewAgentDAO(db),
ActiveSchedule: NewActiveScheduleDAO(db),
ActiveScheduleSession: NewActiveScheduleSessionDAO(db),
Notification: NewNotificationChannelDAO(db),
}
}
@@ -41,15 +43,16 @@ func NewManager(db *gorm.DB) *RepoManager {
// 3. 适用于 outbox 消费处理器这类“基础设施事务 + 业务事务合并”的场景。
func (m *RepoManager) WithTx(tx *gorm.DB) *RepoManager {
return &RepoManager{
db: tx,
Schedule: m.Schedule.WithTx(tx),
Task: m.Task.WithTx(tx),
TaskClass: m.TaskClass.WithTx(tx),
Course: m.Course.WithTx(tx),
User: m.User.WithTx(tx),
Agent: m.Agent.WithTx(tx),
ActiveSchedule: m.ActiveSchedule.WithTx(tx),
Notification: m.Notification.WithTx(tx),
db: tx,
Schedule: m.Schedule.WithTx(tx),
Task: m.Task.WithTx(tx),
TaskClass: m.TaskClass.WithTx(tx),
Course: m.Course.WithTx(tx),
User: m.User.WithTx(tx),
Agent: m.Agent.WithTx(tx),
ActiveSchedule: m.ActiveSchedule.WithTx(tx),
ActiveScheduleSession: m.ActiveScheduleSession.WithTx(tx),
Notification: m.Notification.WithTx(tx),
}
}

View File

@@ -622,6 +622,131 @@ func (d *CacheDAO) DeleteConversationTimelineFromCache(ctx context.Context, user
// Key 设计:
// 1. 使用 smartflow:agent_state 前缀,与现有 key 命名空间隔离;
// 2. 使用 conversationID 作为唯一标识,因为 agent 状态是按会话维度持久化的。
const activeScheduleSessionCacheTTL = 2 * time.Hour
// activeScheduleSessionKey 生成 session_id 维度的主动调度会话缓存 key。
func (d *CacheDAO) activeScheduleSessionKey(sessionID string) string {
return fmt.Sprintf("smartflow:active_schedule_session:s:%s", strings.TrimSpace(sessionID))
}
// activeScheduleSessionConversationKey 生成 user_id + conversation_id 维度的主动调度会话缓存 key。
func (d *CacheDAO) activeScheduleSessionConversationKey(userID int, conversationID string) string {
return fmt.Sprintf("smartflow:active_schedule_session:u:%d:c:%s", userID, strings.TrimSpace(conversationID))
}
// SetActiveScheduleSessionToCache 同步写入主动调度会话缓存。
//
// 步骤化说明:
// 1. 先校验 snapshot 和主键,避免把无效会话写进 Redis
// 2. 再把同一份快照写入 session_id / conversation_id 两个维度的 key
// 3. 若 conversation_id 还没绑定,只写 session_id key避免生成空路由 key。
func (d *CacheDAO) SetActiveScheduleSessionToCache(ctx context.Context, snapshot *model.ActiveScheduleSessionSnapshot) error {
if d == nil || d.client == nil {
return errors.New("cache dao is not initialized")
}
if snapshot == nil {
return errors.New("active schedule session snapshot is nil")
}
sessionID := strings.TrimSpace(snapshot.SessionID)
if sessionID == "" {
return errors.New("session_id is empty")
}
data, err := json.Marshal(snapshot)
if err != nil {
return fmt.Errorf("marshal active schedule session cache failed: %w", err)
}
pipe := d.client.Pipeline()
pipe.Set(ctx, d.activeScheduleSessionKey(sessionID), data, activeScheduleSessionCacheTTL)
if conversationID := strings.TrimSpace(snapshot.ConversationID); conversationID != "" && snapshot.UserID > 0 {
pipe.Set(ctx, d.activeScheduleSessionConversationKey(snapshot.UserID, conversationID), data, activeScheduleSessionCacheTTL)
}
_, err = pipe.Exec(ctx)
return err
}
// GetActiveScheduleSessionFromCache 按 session_id 读取主动调度会话缓存。
func (d *CacheDAO) GetActiveScheduleSessionFromCache(ctx context.Context, sessionID string) (*model.ActiveScheduleSessionSnapshot, error) {
if d == nil || d.client == nil {
return nil, errors.New("cache dao is not initialized")
}
normalizedSessionID := strings.TrimSpace(sessionID)
if normalizedSessionID == "" {
return nil, errors.New("session_id is empty")
}
raw, err := d.client.Get(ctx, d.activeScheduleSessionKey(normalizedSessionID)).Result()
if errors.Is(err, redis.Nil) {
return nil, nil
}
if err != nil {
return nil, err
}
var snapshot model.ActiveScheduleSessionSnapshot
if err = json.Unmarshal([]byte(raw), &snapshot); err != nil {
return nil, fmt.Errorf("unmarshal active schedule session cache failed: %w", err)
}
return &snapshot, nil
}
// GetActiveScheduleSessionFromConversationCache 按 user_id + conversation_id 读取主动调度会话缓存。
func (d *CacheDAO) GetActiveScheduleSessionFromConversationCache(ctx context.Context, userID int, conversationID string) (*model.ActiveScheduleSessionSnapshot, error) {
if d == nil || d.client == nil {
return nil, errors.New("cache dao is not initialized")
}
if userID <= 0 {
return nil, fmt.Errorf("invalid user_id: %d", userID)
}
normalizedConversationID := strings.TrimSpace(conversationID)
if normalizedConversationID == "" {
return nil, errors.New("conversation_id is empty")
}
raw, err := d.client.Get(ctx, d.activeScheduleSessionConversationKey(userID, normalizedConversationID)).Result()
if errors.Is(err, redis.Nil) {
return nil, nil
}
if err != nil {
return nil, err
}
var snapshot model.ActiveScheduleSessionSnapshot
if err = json.Unmarshal([]byte(raw), &snapshot); err != nil {
return nil, fmt.Errorf("unmarshal active schedule session cache failed: %w", err)
}
return &snapshot, nil
}
// DeleteActiveScheduleSessionFromCache 删除主动调度会话缓存。
//
// 说明:
// 1. 会同时清理 session_id 和 conversation_id 两个维度,避免旧路由缓存残留;
// 2. conversation_id 为空时只清 session_id key
// 3. 删除操作本身幂等,即使 key 不存在也视为成功。
func (d *CacheDAO) DeleteActiveScheduleSessionFromCache(ctx context.Context, sessionID string, userID int, conversationID string) error {
if d == nil || d.client == nil {
return errors.New("cache dao is not initialized")
}
normalizedSessionID := strings.TrimSpace(sessionID)
if normalizedSessionID == "" {
return errors.New("session_id is empty")
}
keys := []string{d.activeScheduleSessionKey(normalizedSessionID)}
if userID > 0 {
if normalizedConversationID := strings.TrimSpace(conversationID); normalizedConversationID != "" {
keys = append(keys, d.activeScheduleSessionConversationKey(userID, normalizedConversationID))
}
}
return d.client.Del(ctx, keys...).Err()
}
func (d *CacheDAO) agentStateKey(conversationID string) string {
return fmt.Sprintf("smartflow:agent_state:%s", conversationID)
}

View File

@@ -6,6 +6,7 @@ github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZ
github.com/avast/retry-go v3.0.0+incompatible/go.mod h1:XtSnn+n/sHqQIpZ10K1qAevBhOOCWBLXXy3hyiqqBrY=
github.com/bahlo/generic-list-go v0.2.0 h1:5sz/EEAK+ls5wF+NeqDpk5+iNdMDXrh3z3nPnH1Wvgk=
github.com/bahlo/generic-list-go v0.2.0/go.mod h1:2KvAjgMlE5NNynlg/5iLrrCCZ2+5xWbdbCW3pNTGyYg=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
@@ -46,6 +47,7 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k=
github.com/evanphx/json-patch v0.5.2/go.mod h1:ZWS5hhDbVDyob71nXKNL0+PWn6ToqBHMikGIFbs31qQ=
github.com/francoispqt/gojay v1.2.13/go.mod h1:ehT5mTG4ua4581f1++1WLG0vPdaA9HaiDsoyrBGkyDY=
github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
@@ -146,6 +148,7 @@ github.com/mattn/go-colorable v0.1.2 h1:/bC9yWikZXAL9uJdulbSfyVNIR3n3trXl+v8+1sx
github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
github.com/meguminnnnnnnnn/go-openai v0.1.1 h1:u/IMMgrj/d617Dh/8BKAwlcstD74ynOJzCtVl+y8xAs=
github.com/meguminnnnnnnnn/go-openai v0.1.1/go.mod h1:qs96ysDmxhE4BZoU45I43zcyfnaYxU3X+aRzLko/htY=
github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b h1:j7+1HpAFS1zy5+Q4qx1fWh90gTKwiN4QCGoY9TWyyO4=
@@ -175,7 +178,11 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho=
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc=
github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
github.com/quic-go/quic-go v0.54.0 h1:6s1YB9QotYI6Ospeiguknbp2Znb/jZYjZLRXn9kMQBg=
@@ -302,6 +309,7 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
golang.org/x/telemetry v0.0.0-20250710130107-8d8967aff50b/go.mod h1:4ZwOYna0/zsOKwuR5X/m0QFOJpSZvAxFfkQT+Erd9D4=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
@@ -367,7 +375,9 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gorm.io/driver/mysql v1.6.0 h1:eNbLmNTpPpTOVZi8MMxCi2aaIm0ZpInbORNXDwyLGvg=
gorm.io/driver/mysql v1.6.0/go.mod h1:D/oCC2GWK3M/dqoLxnOlaNKmXz8WNTfcS9y5ovaSqKo=
gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg=
gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=

View File

@@ -28,6 +28,7 @@ func autoMigrateModels(db *gorm.DB) error {
&model.UserNotificationChannel{},
&model.AgentOutboxMessage{},
&model.AgentScheduleState{},
&model.ActiveScheduleSession{},
&model.AgentStateSnapshotRecord{},
&model.MemoryItem{},
&model.MemoryJob{},

View File

@@ -0,0 +1,82 @@
package model
import "time"
const (
// ActiveScheduleSessionStatusWaitingUserReply 表示当前会话正在等待用户补充信息,后端应拦截普通聊天。
ActiveScheduleSessionStatusWaitingUserReply = "waiting_user_reply"
// ActiveScheduleSessionStatusRerunning 表示用户已回复,主动调度图正在重跑,后端仍需拦截普通聊天。
ActiveScheduleSessionStatusRerunning = "rerunning"
// ActiveScheduleSessionStatusReadyPreview 表示已生成可展示预览,当前会话可以释放回普通聊天。
ActiveScheduleSessionStatusReadyPreview = "ready_preview"
// ActiveScheduleSessionStatusApplied 表示用户已确认应用,主动调度会话已经收口。
ActiveScheduleSessionStatusApplied = "applied"
// ActiveScheduleSessionStatusIgnored 表示用户明确忽略本次建议。
ActiveScheduleSessionStatusIgnored = "ignored"
// ActiveScheduleSessionStatusExpired 表示会话已过期,不再承担路由管辖权。
ActiveScheduleSessionStatusExpired = "expired"
// ActiveScheduleSessionStatusFailed 表示会话在绑定、重跑或写回过程中失败。
ActiveScheduleSessionStatusFailed = "failed"
)
// ActiveScheduleSession 是“主动调度会话路由桥”的持久化模型。
//
// 职责边界:
// 1. 只保存会话级路由权与轻量状态,不承载 preview 主表的完整业务内容;
// 2. conversation_id 允许在通知前为空,等站内会话绑定完成后再写入;
// 3. state_json 只存轻量状态,避免把重对象和历史消息继续塞进 session 表。
type ActiveScheduleSession struct {
SessionID string `gorm:"column:session_id;type:varchar(64);primaryKey"`
// 1. user_id + conversation_id 用于在聊天入口侧定位当前管辖中的主动调度会话。
// 2. conversation_id 允许为空,因此这里使用可空列,方便先建 session 再绑定会话。
UserID int `gorm:"column:user_id;not null;index:idx_active_schedule_sessions_user_conv,priority:1;index:idx_active_schedule_sessions_user_status_updated,priority:1"`
ConversationID *string `gorm:"column:conversation_id;type:varchar(128);index:idx_active_schedule_sessions_user_conv,priority:2;index:idx_active_schedule_sessions_conversation_status_updated,priority:1"`
// 3. trigger_id / current_preview_id 分别串起触发源与当前预览,方便后续审计和回放。
TriggerID string `gorm:"column:trigger_id;type:varchar(64);not null;index:idx_active_schedule_sessions_trigger_id"`
CurrentPreviewID *string `gorm:"column:current_preview_id;type:varchar(64);index:idx_active_schedule_sessions_preview_id"`
Status string `gorm:"column:status;type:varchar(32);not null;default:'waiting_user_reply';index:idx_active_schedule_sessions_user_status_updated,priority:2;index:idx_active_schedule_sessions_status_updated,priority:1;index:idx_active_schedule_sessions_conversation_status_updated,priority:2"`
StateJSON string `gorm:"column:state_json;type:json;not null"`
CreatedAt time.Time `gorm:"column:created_at;autoCreateTime"`
UpdatedAt time.Time `gorm:"column:updated_at;autoUpdateTime;index:idx_active_schedule_sessions_user_status_updated,priority:3;index:idx_active_schedule_sessions_status_updated,priority:2;index:idx_active_schedule_sessions_conversation_status_updated,priority:3"`
}
// TableName 返回主动调度会话表名。
func (ActiveScheduleSession) TableName() string {
return "active_schedule_sessions"
}
// ActiveScheduleSessionState 是 session 表内 state_json 对应的轻量状态。
//
// 职责边界:
// 1. 这里只放“路由和补信息闭环”需要的少量字段;
// 2. 不承载完整 preview、正文历史或大块工具结果
// 3. 便于 cache / DAO 之间直接复用同一份 JSON 语义。
type ActiveScheduleSessionState struct {
PendingQuestion string `json:"pending_question,omitempty"`
MissingInfo []string `json:"missing_info,omitempty"`
LastCandidateID string `json:"last_candidate_id,omitempty"`
LastNotificationID string `json:"last_notification_id,omitempty"`
ExpiresAt *time.Time `json:"expires_at,omitempty"`
FailedReason string `json:"failed_reason,omitempty"`
}
// ActiveScheduleSessionSnapshot 是 service、DAO、cache 之间共享的会话快照 DTO。
//
// 职责边界:
// 1. 负责在三层之间传递强类型会话状态;
// 2. 不负责业务决策,不负责拦截判定;
// 3. DAO 再把它拆成数据库列和 state_jsoncache 则直接按 JSON 存取。
type ActiveScheduleSessionSnapshot struct {
SessionID string
UserID int
ConversationID string
TriggerID string
CurrentPreviewID string
Status string
State ActiveScheduleSessionState
CreatedAt time.Time
UpdatedAt time.Time
}

View File

@@ -48,19 +48,40 @@ type Task struct {
EstimatedSections int `gorm:"column:estimated_sections;not null;default:1"`
}
// NormalizeEstimatedSections 将预计节数收敛到 MVP 允许范围。
//
// 职责边界:
// 1. 只处理默认值与越界收敛,不判断业务优先级,也不关心调用方来源;
// 2. nil、0、负数统一回退到 1超过 4 的值收敛到 4保证写库与读回口径一致。
func NormalizeEstimatedSections(raw *int) int {
if raw == nil {
return 1
}
value := *raw
if value < 1 {
return 1
}
if value > 4 {
return 4
}
return value
}
type UserAddTaskResponse struct {
ID int `json:"id"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
DeadlineAt *time.Time `json:"deadline_at"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
ID int `json:"id"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
EstimatedSections int `json:"estimated_sections"`
DeadlineAt *time.Time `json:"deadline_at"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
}
type UserAddTaskRequest struct {
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
DeadlineAt *time.Time `json:"deadline_at"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
EstimatedSections int `json:"estimated_sections"`
DeadlineAt *time.Time `json:"deadline_at"`
}
// UserCompleteTaskRequest 是"标记任务完成"接口的请求体。
@@ -114,6 +135,7 @@ type GetUserTaskResp struct {
UserID int `json:"user_id"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
EstimatedSections int `json:"estimated_sections"`
Status string `json:"status"`
Deadline string `json:"deadline"`
IsCompleted bool `json:"is_completed"`

View File

@@ -108,7 +108,7 @@ type AgentGraphDeps struct {
// 2. 这里只保留“创建任务 / 查询任务”两类轻量能力,避免再回退到已下线的孤立工具链。
type QuickTaskDeps struct {
// CreateTask 创建一条四象限任务,返回 task_id。
CreateTask func(userID int, title string, priorityGroup int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (taskID int, err error)
CreateTask func(userID int, title string, priorityGroup int, estimatedSections int, deadlineAt *time.Time, urgencyThresholdAt *time.Time) (taskID int, err error)
// QueryTasks 按条件查询用户任务列表。
QueryTasks func(ctx context.Context, userID int, params TaskQueryParams) ([]TaskQueryResult, error)
}

View File

@@ -26,10 +26,11 @@ type TaskQueryParams struct {
// 2. 结果既可用于 quick_task 节点文本回复,也可供 service 装配其他轻量输出;
// 3. 不负责序列化策略和文案渲染。
type TaskQueryResult struct {
ID int `json:"id"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
PriorityLabel string `json:"priority_label"`
IsCompleted bool `json:"is_completed"`
DeadlineAt string `json:"deadline_at,omitempty"`
ID int `json:"id"`
Title string `json:"title"`
PriorityGroup int `json:"priority_group"`
EstimatedSections int `json:"estimated_sections"`
PriorityLabel string `json:"priority_label"`
IsCompleted bool `json:"is_completed"`
DeadlineAt string `json:"deadline_at,omitempty"`
}

View File

@@ -20,6 +20,7 @@ type TaskQueryTaskRecord struct {
ID int
Title string
PriorityGroup int
EstimatedSections int
IsCompleted bool
DeadlineAt *time.Time
UrgencyThresholdAt *time.Time

View File

@@ -9,6 +9,7 @@ import (
"time"
infrallm "github.com/LoveLosita/smartflow/backend/infra/llm"
taskmodel "github.com/LoveLosita/smartflow/backend/model"
newagentmodel "github.com/LoveLosita/smartflow/backend/newAgent/model"
newagentprompt "github.com/LoveLosita/smartflow/backend/newAgent/prompt"
newagentrouter "github.com/LoveLosita/smartflow/backend/newAgent/router"
@@ -41,6 +42,7 @@ type quickTaskDecision struct {
Title string `json:"title,omitempty"`
DeadlineAt string `json:"deadline_at,omitempty"`
PriorityGroup *int `json:"priority_group,omitempty"`
EstimatedSections *int `json:"estimated_sections,omitempty"`
UrgencyThresholdAt string `json:"urgency_threshold_at,omitempty"`
TaskID *int `json:"task_id,omitempty"`
@@ -137,8 +139,8 @@ func RunQuickTaskNode(ctx context.Context, input QuickTaskNodeInput) error {
}
break
}
log.Printf("[DEBUG] quick_task: 解析结果 chat=%s action=%s title=%s deadline_at=%s priority_group=%v urgency_threshold_at=%q",
flowState.ConversationID, decision.Action, decision.Title, decision.DeadlineAt, decision.PriorityGroup, decision.UrgencyThresholdAt)
log.Printf("[DEBUG] quick_task: 解析结果 chat=%s action=%s title=%s deadline_at=%s priority_group=%v estimated_sections=%v urgency_threshold_at=%q",
flowState.ConversationID, decision.Action, decision.Title, decision.DeadlineAt, decision.PriorityGroup, decision.EstimatedSections, decision.UrgencyThresholdAt)
// 阶段二:流式推送标签后正文。
if visible != "" {
@@ -266,6 +268,7 @@ func handleQuickTaskCreate(
if priorityGroup == 0 {
priorityGroup = quickNoteFallbackPriority(deadline)
}
estimatedSections := taskmodel.NormalizeEstimatedSections(decision.EstimatedSections)
var urgencyThreshold *time.Time
if raw := strings.TrimSpace(decision.UrgencyThresholdAt); raw != "" {
@@ -280,9 +283,9 @@ func handleQuickTaskCreate(
urgencyThreshold = &fallback
}
log.Printf("[DEBUG] quick_task: CreateTask 参数 chat=%s title=%s priorityGroup=%d deadline=%v urgencyThreshold=%v urgency_raw=%q",
flowState.ConversationID, title, priorityGroup, deadline, urgencyThreshold, decision.UrgencyThresholdAt)
taskID, err := input.QuickTaskDeps.CreateTask(flowState.UserID, title, priorityGroup, deadline, urgencyThreshold)
log.Printf("[DEBUG] quick_task: CreateTask 参数 chat=%s title=%s priorityGroup=%d estimatedSections=%d deadline=%v urgencyThreshold=%v urgency_raw=%q estimated_raw=%v",
flowState.ConversationID, title, priorityGroup, estimatedSections, deadline, urgencyThreshold, decision.UrgencyThresholdAt, decision.EstimatedSections)
taskID, err := input.QuickTaskDeps.CreateTask(flowState.UserID, title, priorityGroup, estimatedSections, deadline, urgencyThreshold)
if err != nil {
return quickTaskActionResult{AssistantText: fmt.Sprintf("记录失败了(%s稍后再试试", err)}
}
@@ -290,7 +293,7 @@ func handleQuickTaskCreate(
flowState.UsedQuickNote = true
return quickTaskActionResult{
AssistantText: "已帮你记下这条任务。",
BusinessCard: buildTaskRecordBusinessCard(taskID, title, priorityGroup, deadline, urgencyThreshold),
BusinessCard: buildTaskRecordBusinessCard(taskID, title, priorityGroup, estimatedSections, deadline, urgencyThreshold),
}
}
@@ -351,13 +354,14 @@ func handleQuickTaskQuery(
}
}
func buildTaskRecordBusinessCard(taskID int, title string, priorityGroup int, deadline *time.Time, urgencyThreshold *time.Time) *newagentstream.StreamBusinessCardExtra {
func buildTaskRecordBusinessCard(taskID int, title string, priorityGroup int, estimatedSections int, deadline *time.Time, urgencyThreshold *time.Time) *newagentstream.StreamBusinessCardExtra {
data := map[string]any{
"id": taskID,
"title": strings.TrimSpace(title),
"priority_group": priorityGroup,
"priority_label": newagentshared.PriorityLabelCN(priorityGroup),
"status": "todo",
"id": taskID,
"title": strings.TrimSpace(title),
"priority_group": priorityGroup,
"estimated_sections": estimatedSections,
"priority_label": newagentshared.PriorityLabelCN(priorityGroup),
"status": "todo",
}
if formatted := formatQuickTaskTime(deadline); formatted != "" {
data["deadline_at"] = formatted
@@ -383,11 +387,12 @@ func buildTaskQueryBusinessCard(params newagentmodel.TaskQueryParams, results []
taskItems := make([]map[string]any, 0, len(results))
for _, task := range results {
item := map[string]any{
"id": task.ID,
"title": strings.TrimSpace(task.Title),
"priority_group": task.PriorityGroup,
"priority_label": newagentshared.PriorityLabelCN(task.PriorityGroup),
"is_completed": task.IsCompleted,
"id": task.ID,
"title": strings.TrimSpace(task.Title),
"priority_group": task.PriorityGroup,
"estimated_sections": task.EstimatedSections,
"priority_label": newagentshared.PriorityLabelCN(task.PriorityGroup),
"is_completed": task.IsCompleted,
}
if deadline := strings.TrimSpace(task.DeadlineAt); deadline != "" {
item["deadline_at"] = deadline

View File

@@ -22,7 +22,7 @@ const quickTaskSystemPrompt = `
JSON 字段说明:
- action只能是 create / query / ask
- create 时title 必填deadline_at 必填priority_group 必填,范围 1-4urgency_threshold_at 满足条件时填写,条件在下面
- create 时title 必填deadline_at 必填priority_group 必填,范围 1-4estimated_sections 必填,范围 1-4不确定默认 1urgency_threshold_at 满足条件时填写,条件在下面
- query 时quadrant 可选 1-4keyword 可选limit 可选deadline_after/deadline_before 可选(用于截止时间窗口筛选)
- ask 时question 必填
@@ -37,9 +37,9 @@ JSON 字段说明:
示例:
<SMARTFLOW_DECISION>{"action":"create","title":"明天开会","deadline_at":"明天下午3点"}</SMARTFLOW_DECISION>
<SMARTFLOW_DECISION>{"action":"create","title":"明天开会","deadline_at":"明天下午3点","estimated_sections":1}</SMARTFLOW_DECISION>
好的,我来帮你记一下。
<SMARTFLOW_DECISION>{"action":"create","title":"下周交报告","deadline_at":"下周五 18:00","priority_group":2,"urgency_threshold_at":"下周四 09:00"}</SMARTFLOW_DECISION>
<SMARTFLOW_DECISION>{"action":"create","title":"下周交报告","deadline_at":"下周五 18:00","priority_group":2,"estimated_sections":2,"urgency_threshold_at":"下周四 09:00"}</SMARTFLOW_DECISION>
好的,我也帮你记一下。
<SMARTFLOW_DECISION>{"action":"query","limit":5}</SMARTFLOW_DECISION>

View File

@@ -154,7 +154,7 @@ func (s *ChannelService) TestFeishuWebhook(ctx context.Context, userID int) (Tes
TriggerType: "manual_test",
TargetType: "notification_channel",
TargetID: 0,
TargetURL: "/schedule-adjust/asp_test_webhook",
TargetURL: "/assistant/00000000-0000-0000-0000-000000000000",
MessageText: "这是一条 SmartFlow 飞书 Webhook 测试消息。",
TraceID: traceID,
AttemptCount: 1,

View File

@@ -334,6 +334,11 @@ var ( //请求相关的响应
Info: "schedule plan preview not found",
}
ConversationNotFound = Response{ //会话不存在或不属于当前用户
Status: "40401",
Info: "conversation not found",
}
MissingConversationID = Response{ //确认/恢复请求缺少会话ID
Status: "40054",
Info: "conversation_id is required when confirm_action is present",

View File

@@ -16,10 +16,19 @@ type AgentService = agentsvc.AgentService
// NewAgentService 是迁移期兼容构造函数。
//
// 说明:
// 1) 外部调用签名不变,新增排程依赖通过可选方式注入(见 NewAgentServiceWithSchedule
// 2) 真实构造逻辑已下沉到 service/agentsvc 包。
func NewAgentService(aiHub *inits.AIHub, repo *dao.AgentDAO, taskRepo *dao.TaskDAO, cacheDAO *dao.CacheDAO, agentRedis *dao.AgentCache, eventPublisher outboxinfra.EventPublisher) *AgentService {
return agentsvc.NewAgentService(aiHub, repo, taskRepo, cacheDAO, agentRedis, eventPublisher)
// 1) 继续保留 service 层入口形式,避免 api/cmd 侧直接感知 agentsvc 包路径
// 2) 主动调度 session DAO 也在这里显式透传,避免聊天入口再去回查全局单例;
// 3) 真实构造逻辑已下沉到 service/agentsvc 包。
func NewAgentService(
aiHub *inits.AIHub,
repo *dao.AgentDAO,
taskRepo *dao.TaskDAO,
cacheDAO *dao.CacheDAO,
agentRedis *dao.AgentCache,
activeSessionDAO *dao.ActiveScheduleSessionDAO,
eventPublisher outboxinfra.EventPublisher,
) *AgentService {
return agentsvc.NewAgentService(aiHub, repo, taskRepo, cacheDAO, agentRedis, activeSessionDAO, eventPublisher)
}
// NewAgentServiceWithSchedule 在基础 AgentService 上注入排程依赖。
@@ -27,18 +36,19 @@ func NewAgentService(aiHub *inits.AIHub, repo *dao.AgentDAO, taskRepo *dao.TaskD
// 设计目的:
// 1) 通过函数注入避免 agentsvc 包直接依赖 service 层的 ScheduleService
// 2) 排程依赖为可选:未注入时排程路由自动回退到普通聊天;
// 3) 保持 NewAgentService 签名不变,向下兼容
// 3) 主动调度 session DAO 仍沿用统一构造注入,避免排程分支自己拼装仓储
func NewAgentServiceWithSchedule(
aiHub *inits.AIHub,
repo *dao.AgentDAO,
taskRepo *dao.TaskDAO,
cacheDAO *dao.CacheDAO,
agentRedis *dao.AgentCache,
activeSessionDAO *dao.ActiveScheduleSessionDAO,
eventPublisher outboxinfra.EventPublisher,
scheduleSvc *ScheduleService,
taskSvc *TaskService,
) *AgentService {
svc := agentsvc.NewAgentService(aiHub, repo, taskRepo, cacheDAO, agentRedis, eventPublisher)
svc := agentsvc.NewAgentService(aiHub, repo, taskRepo, cacheDAO, agentRedis, activeSessionDAO, eventPublisher)
// 注入排程依赖:将 service 层方法包装为函数闭包,避免循环依赖。
if scheduleSvc != nil {

View File

@@ -26,12 +26,13 @@ import (
)
type AgentService struct {
AIHub *inits.AIHub
repo *dao.AgentDAO
taskRepo *dao.TaskDAO
cacheDAO *dao.CacheDAO
agentCache *dao.AgentCache
eventPublisher outboxinfra.EventPublisher
AIHub *inits.AIHub
repo *dao.AgentDAO
taskRepo *dao.TaskDAO
cacheDAO *dao.CacheDAO
agentCache *dao.AgentCache
activeScheduleSessionDAO *dao.ActiveScheduleSessionDAO
eventPublisher outboxinfra.EventPublisher
// ── 排程计划依赖(函数注入,避免 service 包循环依赖)──
@@ -66,24 +67,34 @@ type AgentService struct {
memoryCfg memorymodel.Config
memoryObserver memoryobserve.Observer
memoryMetrics memoryobserve.MetricsRecorder
activeRerunFunc ActiveScheduleSessionRerunFunc
}
// NewAgentService 构造 AgentService。
// 这里通过依赖注入把“模型、仓储、缓存、异步持久化通道”统一交给服务层管理,
// 便于后续在单测中替换实现,或在启动流程中按环境切换配置。
func NewAgentService(aiHub *inits.AIHub, repo *dao.AgentDAO, taskRepo *dao.TaskDAO, cacheDAO *dao.CacheDAO, agentRedis *dao.AgentCache, eventPublisher outboxinfra.EventPublisher) *AgentService {
func NewAgentService(
aiHub *inits.AIHub,
repo *dao.AgentDAO,
taskRepo *dao.TaskDAO,
cacheDAO *dao.CacheDAO,
agentRedis *dao.AgentCache,
activeSessionDAO *dao.ActiveScheduleSessionDAO,
eventPublisher outboxinfra.EventPublisher,
) *AgentService {
// 全局注册一次 token 采集 callback
// 1. 只注册一次,避免重复处理;
// 2. 只有带 RequestTokenMeter 的请求上下文才会真正累加。
ensureTokenMeterCallbackRegistered()
return &AgentService{
AIHub: aiHub,
repo: repo,
taskRepo: taskRepo,
cacheDAO: cacheDAO,
agentCache: agentRedis,
eventPublisher: eventPublisher,
AIHub: aiHub,
repo: repo,
taskRepo: taskRepo,
cacheDAO: cacheDAO,
agentCache: agentRedis,
activeScheduleSessionDAO: activeSessionDAO,
eventPublisher: eventPublisher,
}
}

View File

@@ -0,0 +1,323 @@
package agentsvc
import (
"context"
"errors"
"log"
"strings"
"time"
"github.com/LoveLosita/smartflow/backend/model"
newagentstream "github.com/LoveLosita/smartflow/backend/newAgent/stream"
"github.com/cloudwego/eino/schema"
)
// ActiveScheduleSessionRerunFunc 表示主动调度 session 被聊天入口接管后,如何同步推进 rerun。
//
// 职责边界:
// 1. 只负责把“当前 session + 用户回复”推进为新的主动调度结果;
// 2. 不负责决定 session 何时创建,也不负责通知投递;
// 3. 返回的结果只面向聊天入口的可见消息和 session 状态回写。
type ActiveScheduleSessionRerunFunc func(
ctx context.Context,
session *model.ActiveScheduleSessionSnapshot,
userMessage string,
traceID string,
requestStart time.Time,
) (*ActiveScheduleSessionRerunResult, error)
// ActiveScheduleSessionRerunResult 是主动调度 rerun 的最小返回结果。
//
// 职责边界:
// 1. 只承载聊天入口需要回写的可见消息、业务卡片和 session 状态;
// 2. 不直接暴露 DAO 行,也不承载 worker / notification 的副作用;
// 3. AssistantText 为空时,调用方可降级为使用卡片摘要。
type ActiveScheduleSessionRerunResult struct {
AssistantText string
BusinessCard *newagentstream.StreamBusinessCardExtra
SessionState model.ActiveScheduleSessionState
SessionStatus string
PreviewID string
}
// SetActiveScheduleSessionRerunFunc 注入主动调度 rerun 入口。
func (s *AgentService) SetActiveScheduleSessionRerunFunc(fn ActiveScheduleSessionRerunFunc) {
s.activeRerunFunc = fn
}
// loadActiveScheduleSessionByConversation 尽量从缓存 + 数据库读取当前会话的主动调度 session。
//
// 步骤化说明:
// 1. 先读 Redis 热缓存,命中则直接返回;
// 2. 缓存未命中再回源数据库,避免把 session 状态逻辑绑死在缓存上;
// 3. 回源成功后尽力回填缓存,减少下一轮聊天入口的 DB 压力。
func (s *AgentService) loadActiveScheduleSessionByConversation(ctx context.Context, userID int, chatID string) (*model.ActiveScheduleSessionSnapshot, error) {
if s == nil || s.activeScheduleSessionDAO == nil {
return nil, nil
}
normalizedChatID := strings.TrimSpace(chatID)
if userID <= 0 || normalizedChatID == "" {
return nil, nil
}
if s.cacheDAO != nil {
cached, err := s.cacheDAO.GetActiveScheduleSessionFromConversationCache(ctx, userID, normalizedChatID)
if err != nil {
log.Printf("读取主动调度 session 缓存失败 user=%d chat=%s err=%v", userID, normalizedChatID, err)
} else if cached != nil {
return cached, nil
}
}
row, err := s.activeScheduleSessionDAO.GetActiveScheduleSessionByConversationID(ctx, userID, normalizedChatID)
if err != nil || row == nil {
return nil, err
}
if s.cacheDAO != nil {
if cacheErr := s.cacheDAO.SetActiveScheduleSessionToCache(ctx, row); cacheErr != nil {
log.Printf("回填主动调度 session 缓存失败 user=%d chat=%s err=%v", userID, normalizedChatID, cacheErr)
}
}
return row, nil
}
// persistActiveScheduleSessionBestEffort 负责把主动调度 session 的最新状态同步回 MySQL 和 Redis。
//
// 职责边界:
// 1. MySQL 是最终真相,先写表再回填缓存;
// 2. 缓存失败只记日志,不影响主流程;
// 3. 调用方需要先把 snapshot 改成最终状态,再交给这里落盘。
func (s *AgentService) persistActiveScheduleSessionBestEffort(ctx context.Context, snapshot *model.ActiveScheduleSessionSnapshot) error {
if s == nil || s.activeScheduleSessionDAO == nil || snapshot == nil {
return nil
}
if strings.TrimSpace(snapshot.SessionID) == "" {
return errors.New("active schedule session_id 不能为空")
}
if err := s.activeScheduleSessionDAO.UpsertActiveScheduleSession(ctx, snapshot); err != nil {
return err
}
// 1. 重新读取一遍,拿到数据库侧最终落表后的标准快照,减少缓存和 DB 的口径漂移。
// 2. 如果重读失败,也不影响主链路返回,只要主表已成功写入即可。
normalized, err := s.activeScheduleSessionDAO.GetActiveScheduleSessionBySessionID(ctx, snapshot.SessionID)
if err == nil && normalized != nil {
snapshot = normalized
}
if s.cacheDAO != nil {
if cacheErr := s.cacheDAO.SetActiveScheduleSessionToCache(ctx, snapshot); cacheErr != nil {
log.Printf("回填主动调度 session 缓存失败 session=%s err=%v", snapshot.SessionID, cacheErr)
}
}
return nil
}
// handleActiveScheduleSessionChat 处理被主动调度 session 占管的聊天入口。
//
// 步骤化说明:
// 1. 先读 session判断当前 conversation 是否仍在 waiting_user_reply / rerunning 占管期;
// 2. 占管期间先把用户消息写入历史和时间线,保证会话内容不丢失;
// 3. waiting_user_reply 进入 rerunning并同步调用主动调度 rerun
// 4. rerunning 则只提示“正在重跑”,避免同一 conversation 被并发重复推进;
// 5. 终态或非占管态直接放行普通 newAgent。
func (s *AgentService) handleActiveScheduleSessionChat(
ctx context.Context,
userMessage string,
traceID string,
requestStart time.Time,
userID int,
chatID string,
resolvedModelName string,
outChan chan<- string,
errChan chan error,
) (bool, error) {
session, err := s.loadActiveScheduleSessionByConversation(ctx, userID, chatID)
if err != nil {
return false, err
}
if session == nil || !isActiveScheduleSessionBlockingStatus(session.Status) {
return false, nil
}
trimmedMessage := strings.TrimSpace(userMessage)
if trimmedMessage != "" {
// 1. 主动调度占管期间,用户每次回复仍然要进入正常会话历史。
// 2. 这样后续刷新聊天页时,用户可见消息、时间线和 session 状态不会彼此脱节。
if err := s.persistNewAgentConversationMessage(ctx, userID, chatID, schema.UserMessage(trimmedMessage), 0); err != nil {
return true, err
}
}
switch session.Status {
case model.ActiveScheduleSessionStatusWaitingUserReply:
if trimmedMessage == "" {
assistantText := strings.TrimSpace(session.State.PendingQuestion)
if assistantText == "" {
assistantText = "请先补充主动调度需要的关键信息。"
}
if err := s.persistNewAgentConversationMessage(ctx, userID, chatID, schema.AssistantMessage(assistantText, nil), 0); err != nil {
return true, err
}
emitActiveScheduleAssistantChunk(outChan, traceID, resolvedModelName, requestStart, assistantText, nil)
return true, nil
}
// 1. 收到用户补充信息后,先把 session 切成 rerunning避免并发请求继续按旧状态走普通聊天。
// 2. 这个阶段只是状态切换,不代表 graph 已经完成。
session.Status = model.ActiveScheduleSessionStatusRerunning
if err := s.persistActiveScheduleSessionBestEffort(ctx, session); err != nil {
return true, err
}
return true, s.runActiveScheduleSessionRerun(ctx, session, trimmedMessage, traceID, requestStart, resolvedModelName, outChan, errChan)
case model.ActiveScheduleSessionStatusRerunning:
// 1. rerunning 是占管中的过渡态,说明当前会话已经在重跑或刚开始重跑。
// 2. 这里不再触发第二次 rerun只给用户一个可见的等待提示。
if trimmedMessage != "" {
assistantText := "主动调度正在重新生成建议,请稍后再试。"
if err := s.persistNewAgentConversationMessage(ctx, userID, chatID, schema.AssistantMessage(assistantText, nil), 0); err != nil {
return true, err
}
emitActiveScheduleAssistantChunk(outChan, traceID, resolvedModelName, requestStart, assistantText, nil)
}
return true, nil
default:
return false, nil
}
}
// runActiveScheduleSessionRerun 负责把 waiting_user_reply 的用户补充同步推进成新的主动调度结果。
//
// 职责边界:
// 1. 只负责聊天入口的最小编排,不复制 worker / notification 链路;
// 2. 成功时把新 preview / ask_user / close 的结果写回 session + timeline
// 3. 失败时把 session 标成 failed方便后续排障。
func (s *AgentService) runActiveScheduleSessionRerun(
ctx context.Context,
session *model.ActiveScheduleSessionSnapshot,
userMessage string,
traceID string,
requestStart time.Time,
resolvedModelName string,
outChan chan<- string,
errChan chan error,
) error {
if s == nil || s.activeRerunFunc == nil {
return errors.New("主动调度 rerun 未接入")
}
if session == nil {
return errors.New("active schedule session 不能为空")
}
result, err := s.activeRerunFunc(ctx, session, userMessage, traceID, requestStart)
if err != nil {
session.Status = model.ActiveScheduleSessionStatusFailed
session.State.FailedReason = strings.TrimSpace(err.Error())
_ = s.persistActiveScheduleSessionBestEffort(ctx, session)
return err
}
if result == nil {
result = &ActiveScheduleSessionRerunResult{}
}
finalStatus := strings.TrimSpace(result.SessionStatus)
if finalStatus == "" {
if result.BusinessCard != nil {
finalStatus = model.ActiveScheduleSessionStatusReadyPreview
} else {
finalStatus = model.ActiveScheduleSessionStatusWaitingUserReply
}
}
session.Status = finalStatus
session.State = result.SessionState
if strings.TrimSpace(result.PreviewID) != "" {
session.CurrentPreviewID = strings.TrimSpace(result.PreviewID)
} else if session.Status != model.ActiveScheduleSessionStatusReadyPreview {
session.CurrentPreviewID = ""
}
if session.Status == model.ActiveScheduleSessionStatusReadyPreview {
session.State.PendingQuestion = ""
session.State.MissingInfo = nil
session.State.FailedReason = ""
}
if err := s.persistActiveScheduleSessionBestEffort(ctx, session); err != nil {
return err
}
assistantText := strings.TrimSpace(result.AssistantText)
if assistantText == "" && result.BusinessCard != nil {
assistantText = strings.TrimSpace(result.BusinessCard.Summary)
}
if assistantText == "" {
assistantText = "主动调度建议已更新。"
}
// 1. 把新结果写进 conversation history保证刷新后仍然能看到 rerun 的正文。
// 2. 再追加业务卡片时间线,前端可以按 timeline 重建主动调度卡片。
if err := s.persistNewAgentConversationMessage(ctx, session.UserID, session.ConversationID, schema.AssistantMessage(assistantText, nil), 0); err != nil {
return err
}
if result.BusinessCard != nil {
if _, err := s.appendConversationTimelineEvent(
ctx,
session.UserID,
session.ConversationID,
model.AgentTimelineKindBusinessCard,
"assistant",
assistantText,
map[string]any{"business_card": result.BusinessCard},
0,
); err != nil {
return err
}
}
emitActiveScheduleAssistantChunk(outChan, traceID, resolvedModelName, requestStart, assistantText, nil)
if result.BusinessCard != nil {
emitActiveScheduleBusinessCardChunk(outChan, session.SessionID, traceID, resolvedModelName, requestStart, result.BusinessCard)
}
return nil
}
func isActiveScheduleSessionBlockingStatus(status string) bool {
switch strings.ToLower(strings.TrimSpace(status)) {
case model.ActiveScheduleSessionStatusWaitingUserReply,
model.ActiveScheduleSessionStatusRerunning:
return true
default:
return false
}
}
func emitActiveScheduleAssistantChunk(outChan chan<- string, traceID string, modelName string, requestStart time.Time, text string, extra *newagentstream.OpenAIChunkExtra) {
payload, err := newagentstream.ToOpenAIAssistantChunkWithExtra(traceID, modelName, requestStart.Unix(), strings.TrimSpace(text), true, extra)
if err != nil {
log.Printf("构造主动调度 assistant chunk 失败 trace=%s err=%v", traceID, err)
return
}
pushChunkNonBlocking(outChan, payload)
}
func emitActiveScheduleBusinessCardChunk(outChan chan<- string, blockID string, traceID string, modelName string, requestStart time.Time, card *newagentstream.StreamBusinessCardExtra) {
if card == nil {
return
}
payload, err := newagentstream.ToOpenAIStreamWithExtra(nil, traceID, modelName, requestStart.Unix(), true, newagentstream.NewBusinessCardExtra(blockID, "active_schedule_session", card))
if err != nil {
log.Printf("构造主动调度 business card chunk 失败 trace=%s err=%v", traceID, err)
return
}
pushChunkNonBlocking(outChan, payload)
}
func pushChunkNonBlocking(outChan chan<- string, payload string) {
if outChan == nil || strings.TrimSpace(payload) == "" {
return
}
select {
case outChan <- payload:
default:
log.Printf("主动调度 SSE 通道已满,丢弃 payload")
}
}

View File

@@ -87,12 +87,21 @@ func (s *AgentService) runNewAgentGraph(
// 3. retry 机制已下线,不再构建重试元数据。
// 4. 从 StateStore 加载或创建 RuntimeState
// 4. 如果当前 conversation 被主动调度 session 占管,先走 session 分支,不进入普通 newAgent
// 这样 waiting_user_reply / rerunning 期间,用户消息会先推动主动调度闭环,而不是误进自由聊天。
if handled, sessionErr := s.handleActiveScheduleSessionChat(requestCtx, userMessage, traceID, requestStart, userID, chatID, resolvedModelName, outChan, errChan); sessionErr != nil {
pushErrNonBlocking(errChan, sessionErr)
return
} else if handled {
return
}
// 5. 从 StateStore 加载或创建 RuntimeState。
// 恢复场景confirm/ask_user同时拿到快照中保存的 ConversationContext
// 其中包含工具调用/结果等中间消息,保证后续 LLM 调用的消息链完整。
runtimeState, savedConversationContext, savedScheduleState, savedOriginalScheduleState := s.loadOrCreateRuntimeState(requestCtx, chatID, userID)
// 5. 构造 ConversationContext。
// 6. 构造 ConversationContext。
// 优先使用快照中恢复的 ConversationContext含工具调用/结果),
// 无快照时从 Redis LLM 历史缓存加载。
var conversationContext *newagentmodel.ConversationContext
@@ -105,17 +114,17 @@ func (s *AgentService) runNewAgentGraph(
} else {
conversationContext = s.loadConversationContext(requestCtx, chatID, userMessage)
}
// 5.1. 在 graph 执行前统一补充与当前输入相关的记忆上下文(预取管线模式)。
// 5.1.1 先读 Redis 预取缓存注入到 ConversationContext再启动后台 goroutine 做完整检索;
// 5.1.2 返回的 channel 传入 Deps供 Execute/Plan 节点在启动前消费最新记忆;
// 5.1.3 检索失败只降级为"本轮不注入记忆",不阻断主链路。
// 6.1. 在 graph 执行前统一补充与当前输入相关的记忆上下文(预取管线模式)。
// 6.1.1 先读 Redis 预取缓存注入到 ConversationContext再启动后台 goroutine 做完整检索;
// 6.1.2 返回的 channel 传入 Deps供 Execute/Plan 节点在启动前消费最新记忆;
// 6.1.3 检索失败只降级为"本轮不注入记忆",不阻断主链路。
memoryFuture := s.injectMemoryContext(requestCtx, conversationContext, userID, chatID, userMessage)
// 5.5 将前端传入的 thinkingMode 写入 CommonState供 ChatNode 及下游节点读取。
// 6.5 将前端传入的 thinkingMode 写入 CommonState供 ChatNode 及下游节点读取。
cs := runtimeState.EnsureCommonState()
cs.ThinkingMode = thinkingMode
// 5.6 若 extra 携带 task_class_ids校验后写入 CommonState仅首轮/尚未设置时生效,跨轮持久化)。
// 6.6 若 extra 携带 task_class_ids校验后写入 CommonState仅首轮/尚未设置时生效,跨轮持久化)。
if taskClassIDs := readAgentExtraIntSlice(extra, "task_class_ids"); len(taskClassIDs) > 0 {
cs := runtimeState.EnsureCommonState()
if len(cs.TaskClassIDs) == 0 {
@@ -135,7 +144,7 @@ func (s *AgentService) runNewAgentGraph(
cs = runtimeState.EnsureCommonState()
// 5.7 先把本轮用户输入落库,确保后续可见 assistant 消息按真实时间线追加。
// 6.7 先把本轮用户输入落库,确保后续可见 assistant 消息按真实时间线追加。
userMsg := schema.UserMessage(userMessage)
if err := s.persistNewAgentConversationMessage(requestCtx, userID, chatID, userMsg, 0); err != nil {
pushErrNonBlocking(errChan, err)
@@ -158,7 +167,7 @@ func (s *AgentService) runNewAgentGraph(
return s.persistNewAgentConversationMessage(persistCtx, userID, chatID, msg, 0)
}
// 6. 构造 AgentGraphRequest。
// 7. 构造 AgentGraphRequest。
var (
confirmAction string
resumeInteractionID string
@@ -175,16 +184,16 @@ func (s *AgentService) runNewAgentGraph(
}
graphRequest.Normalize()
// 7. 适配 LLM clients从 AIHub 的 ark.ChatModel 转换为 newAgent LLM Client
// 7.1 Chat/Deliver 使用 Pro 模型:路由分流、闲聊、交付总结属于标准复杂度。
// 7.2 Plan/Execute 使用 Max 模型:规划和 ReAct 循环需要深度推理能力。
// 8. 适配 LLM clients从 AIHub 的 ark.ChatModel 转换为 newAgent LLM Client
// 8.1 Chat/Deliver 使用 Pro 模型:路由分流、闲聊、交付总结属于标准复杂度。
// 8.2 Plan/Execute 使用 Max 模型:规划和 ReAct 循环需要深度推理能力。
chatClient := infrallm.WrapArkClient(s.AIHub.Pro)
planClient := infrallm.WrapArkClient(s.AIHub.Max)
executeClient := infrallm.WrapArkClient(s.AIHub.Max)
deliverClient := infrallm.WrapArkClient(s.AIHub.Pro)
summaryClient := infrallm.WrapArkClient(s.AIHub.Lite)
// 8. 适配 SSE emitter。
// 9. 适配 SSE emitter。
sseEmitter := newagentstream.NewSSEPayloadEmitter(outChan)
chunkEmitter := newagentstream.NewChunkEmitter(sseEmitter, traceID, resolvedModelName, requestStart.Unix())
chunkEmitter.SetReasoningSummaryFunc(s.makeReasoningSummaryFunc(summaryClient))
@@ -193,7 +202,7 @@ func (s *AgentService) runNewAgentGraph(
s.persistNewAgentTimelineExtraEvent(context.Background(), userID, chatID, extra)
})
// 9. 构造 AgentGraphDeps由 cmd/start.go 注入的依赖)。
// 10. 构造 AgentGraphDeps由 cmd/start.go 注入的依赖)。
deps := newagentmodel.AgentGraphDeps{
ChatClient: chatClient,
PlanClient: planClient,
@@ -214,7 +223,7 @@ func (s *AgentService) runNewAgentGraph(
QuickTaskDeps: s.quickTaskDeps,
}
// 10. 构造 AgentGraphRunInput 并运行 graph。
// 11. 构造 AgentGraphRunInput 并运行 graph。
runInput := newagentmodel.AgentGraphRunInput{
RuntimeState: runtimeState,
ConversationContext: conversationContext,
@@ -240,10 +249,10 @@ func (s *AgentService) runNewAgentGraph(
return
}
// 11. 持久化聊天历史(用户消息 + 助手回复)。
// 12. 持久化聊天历史(用户消息 + 助手回复)。
requestTotalTokens := snapshotRequestTokenMeter(requestCtx).TotalTokens
s.adjustNewAgentRequestTokenUsage(requestCtx, userID, chatID, requestTotalTokens)
// 11.5. 将最终状态快照异步写入 MySQL通过 outbox
// 12.5. 将最终状态快照异步写入 MySQL通过 outbox
// Deliver 节点已将快照保存到 Redis2h TTL此处通过 outbox 异步写入 MySQL 做永久存储。
if finalState != nil {
snapshot := &newagentmodel.AgentStateSnapshot{
@@ -253,7 +262,7 @@ func (s *AgentService) runNewAgentGraph(
eventsvc.PublishAgentStateSnapshot(requestCtx, s.eventPublisher, snapshot, chatID, userID)
}
// 11.6. graph 完成后条件触发记忆抽取。
// 12.6. graph 完成后条件触发记忆抽取。
// 说明:
// 1. 只有本轮未走快捷随口记任务路径时才触发记忆抽取;
// 2. 避免随口记创建的 Task 与记忆系统产生语义冲突。
@@ -269,10 +278,10 @@ func (s *AgentService) runNewAgentGraph(
// 排程预览缓存由 Deliver 节点负责写入(通过注入的 WriteSchedulePreview func
// 保证只有任务真正完成时才写,中断路径不写中间态。
// 12. 发送 OpenAI 兼容的流式结束标记,告知客户端 stream 已完成。
// 13. 发送 OpenAI 兼容的流式结束标记,告知客户端 stream 已完成。
_ = chunkEmitter.EmitDone()
// 13. 异步生成会话标题。
// 14. 异步生成会话标题。
s.ensureConversationTitleAsync(userID, chatID)
}

View File

@@ -67,6 +67,7 @@ func (s *AgentService) QueryTasksForTool(ctx context.Context, req newagentmodel.
ID: task.ID,
Title: task.Title,
PriorityGroup: task.Priority,
EstimatedSections: model.NormalizeEstimatedSections(&task.EstimatedSections),
IsCompleted: task.IsCompleted,
DeadlineAt: task.DeadlineAt,
UrgencyThresholdAt: task.UrgencyThresholdAt,

View File

@@ -54,8 +54,8 @@ func (p FeishuNotificationRequestedPayload) Validate() error {
if targetURL == "" {
return errors.New("target_url 不能为空")
}
if !strings.HasPrefix(targetURL, "/schedule-adjust/") {
return errors.New("target_url 必须是 /schedule-adjust/{preview_id} 站内相对路径")
if !strings.HasPrefix(targetURL, "/assistant/") {
return errors.New("target_url 必须是 /assistant/{conversation_id} 站内相对路径")
}
if strings.Contains(targetURL, "://") || strings.HasPrefix(targetURL, "//") {
return errors.New("target_url 不允许携带外部链接")