diff --git a/cmd/events/main.go b/cmd/events/main.go index a14fef81..39c9c00c 100644 --- a/cmd/events/main.go +++ b/cmd/events/main.go @@ -16,6 +16,7 @@ import ( "chat-agent-system/internal/app/docproject" "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/config" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" "chat-agent-system/internal/events" @@ -75,6 +76,7 @@ func main() { ) _ = disp.Register(events.ConsumerNotify, (&docnotify.EventConsumer{ Notify: notify, Docs: gov, Env: cfg.Env, + Chat: ctxmgr.NewContextManager(pool, nil, ctxmgr.TimeDecayConfig{}), }).HandleLease) _ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{ Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env, diff --git a/internal/app/app.go b/internal/app/app.go index e7a18a86..af8e8ca5 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -196,6 +196,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { traceSvc := traceimpact.Open(pool) traceSvc.SetStageReopener(traceStage{inst: instSvc}) traceSvc.SetDocUnlocker(traceUnlock{gov: parts.Gov}) + bindRecheckChat(traceSvc, parts.Ctx) if instSvc != nil { instSvc.SetRecheckGate(traceRecheck{svc: traceSvc}) } @@ -255,7 +256,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { instSvc.SetAuthorOnlyPolicySource(&sysadminAuthorOnlyPolicyAdapter{svc: rt.SysAdmin}) svc.SetInstanceDutySource(&projectInstanceDutyAdapter{inst: instSvc}) } - rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: parts.Gov} + rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: parts.Gov, Chat: parts.Ctx} rt.DocNotify.SkipWrite = func(ctx context.Context) bool { return rt.WriterLive(ctx, events.ConsumerNotify) } diff --git a/internal/app/docnotify/chat_notice.go b/internal/app/docnotify/chat_notice.go new file mode 100644 index 00000000..7b85c7f3 --- /dev/null +++ b/internal/app/docnotify/chat_notice.go @@ -0,0 +1,45 @@ +package docnotify + +import ( + "context" + "fmt" + + "chat-agent-system/internal/ctxmgr" +) + +// PostProjectNotice writes an assistant message into the actor's latest project session. +// It does not create a session. Missing chat, user, or session is a no-op. +func PostProjectNotice(ctx context.Context, chat *ctxmgr.ContextManager, projectID, userID int64, content string, meta map[string]any) error { + if chat == nil || projectID <= 0 || userID <= 0 || content == "" { + return nil + } + sessions, err := chat.ListChatSessions(ctx, ctxmgr.SessionFilter{UserID: userID, ProjectID: projectID, Limit: 1}) + if err != nil { + return err + } + if len(sessions) == 0 || sessions[0].SessionID == "" { + return nil + } + if meta == nil { + meta = map[string]any{} + } + if _, ok := meta["project_id"]; !ok { + meta["project_id"] = projectID + } + _, err = chat.RecordUserMessageWithID(ctx, sessions[0].SessionID, 0, "assistant", content, "IMPACT_ANALYZE", "", meta) + if err != nil { + return fmt.Errorf("post project notice: %w", err) + } + return nil +} + +func downstreamChatBody(docID, impacted string) string { + switch { + case docID != "" && impacted != "": + return fmt.Sprintf("文档 %s 的变更可能影响 %s。需要时可以说「分析影响」。", docID, impacted) + case docID != "": + return fmt.Sprintf("文档 %s 有变更，下游内容可能受影响。需要时可以说「分析影响」。", docID) + default: + return "有文档变更，下游内容可能受影响。需要时可以说「分析影响」。" + } +} diff --git a/internal/app/docnotify/chat_notice_test.go b/internal/app/docnotify/chat_notice_test.go new file mode 100644 index 00000000..12ef7854 --- /dev/null +++ b/internal/app/docnotify/chat_notice_test.go @@ -0,0 +1,51 @@ +package docnotify + +import ( + "strings" + "testing" + + "chat-agent-system/internal/ctxmgr" +) + +func TestPostProjectNoticeSkipsWithoutSession(t *testing.T) { + chat := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + if err := PostProjectNotice(t.Context(), chat, 7, 3, "上游有变更", nil); err != nil { + t.Fatal(err) + } + history, err := chat.GetChatHistory(t.Context(), "sess-missing", 10, 0) + if err != nil { + t.Fatal(err) + } + if len(history) != 0 { + t.Fatalf("must not create a session, history=%d", len(history)) + } +} + +func TestPostProjectNoticeWritesActorSession(t *testing.T) { + chat := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + if _, err := chat.EnsureChatSession(t.Context(), "sess-p7-u3", 3, 7, "项目对话"); err != nil { + t.Fatal(err) + } + if err := PostProjectNotice(t.Context(), chat, 7, 3, "文档 d 的变更可能影响 down。", map[string]any{ + "interactive_card": map[string]any{"cardType": "RECHECK_NOTICE"}, + }); err != nil { + t.Fatal(err) + } + history, err := chat.GetChatHistory(t.Context(), "sess-p7-u3", 10, 0) + if err != nil { + t.Fatal(err) + } + if len(history) != 1 || history[0].Role != "assistant" { + t.Fatalf("history=%+v", history) + } + if history[0].Metadata["interactive_card"] == nil { + t.Fatalf("metadata=%+v", history[0].Metadata) + } +} + +func TestDownstreamChatBody(t *testing.T) { + got := downstreamChatBody("doc-a", "doc-b") + if !strings.Contains(got, "doc-a") || !strings.Contains(got, "doc-b") || !strings.Contains(got, "分析影响") { + t.Fatalf("body=%q", got) + } +} diff --git a/internal/app/docnotify/consumer.go b/internal/app/docnotify/consumer.go index ebaebb39..8d70d6d8 100644 --- a/internal/app/docnotify/consumer.go +++ b/internal/app/docnotify/consumer.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/events" "chat-agent-system/internal/notification" @@ -14,6 +15,7 @@ import ( type EventConsumer struct { Notify *notification.Service Docs DocumentLookup + Chat *ctxmgr.ContextManager Env string } @@ -57,7 +59,14 @@ func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx return events.ErrQuarantine } p := notification.DocumentEventPayload{ProjectID: projectID, DocID: docID} - return c.Notify.EmitVisibleDocumentEvent(ctx, tx, lease.EventID, kind, title, body, recipient, p) + if err := c.Notify.EmitVisibleDocumentEvent(ctx, tx, lease.EventID, kind, title, body, recipient, p); err != nil { + return err + } + if lease.EventName == events.EventDownstreamImpact && c.Chat != nil { + impacted, _ := payload["impacted_doc_id"].(string) + _ = PostProjectNotice(ctx, c.Chat, projectID, recipient, downstreamChatBody(docID, impacted), nil) + } + return nil } func IntendedNotifyDigest(eventID, kind string, recipient int64) string { diff --git a/internal/app/docnotify/subscriber.go b/internal/app/docnotify/subscriber.go index 1e516c85..15f45dfc 100644 --- a/internal/app/docnotify/subscriber.go +++ b/internal/app/docnotify/subscriber.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/docgov" "chat-agent-system/internal/docrevision" "chat-agent-system/internal/notification" @@ -19,6 +20,7 @@ type DocumentLookup interface { type Subscriber struct { Notify notificationEmitter Docs DocumentLookup + Chat *ctxmgr.ContextManager // SkipWrite is true when the durable notify consumer is live. // Bind stays registered while mode=off; the closure must not dual-write after switch. SkipWrite func(ctx context.Context) bool @@ -88,7 +90,10 @@ func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[s _, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact, "下游文档受影响", "上游文档变更，请复核关联文档。", fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p) - return err + if err != nil { + return err + } + return PostProjectNotice(ctx, s.Chat, projectID, recipient, downstreamChatBody(docID, impacted), nil) case notification.EventDocChangeReportDegraded: if v, ok := payload["report_id"].(int64); ok { p.ReportID = v diff --git a/internal/app/trace_bridge.go b/internal/app/trace_bridge.go index 0b682b5c..409308f7 100644 --- a/internal/app/trace_bridge.go +++ b/internal/app/trace_bridge.go @@ -7,7 +7,9 @@ import ( "strconv" "strings" + "chat-agent-system/internal/app/docnotify" "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/docgov" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/traceimpact" @@ -60,15 +62,54 @@ func (t traceUnlock) ReopenApprovedDocs(ctx context.Context, projectID int64, do type traceChat struct{ svc *traceimpact.Service } -func (t traceChat) AnalyzeImpact(ctx context.Context, projectID, _ int64, text string) (string, error) { +func (t traceChat) AnalyzeImpact(ctx context.Context, projectID, userID int64, text string) (string, error) { if t.svc == nil { return "", fmt.Errorf("trace: unavailable") } + ctx = traceimpact.WithActor(ctx, userID) rep, err := t.svc.Analyze(ctx, projectID, traceimpact.DocRef{DocID: "chat", Markdown: text, Visible: true}, nil) if err != nil { return "", err } - return rep.Summary, nil + return formatImpactReply(rep), nil +} + +func formatImpactReply(rep *traceimpact.Report) string { + if rep == nil || len(rep.Impacts) == 0 { + if rep.Summary != "" { + return rep.Summary + } + return "目前没有已确认的影响。可以说「分析影响」并带上条目编号，例如 UR-001。" + } + var b strings.Builder + if rep.Summary != "" { + b.WriteString(rep.Summary) + b.WriteString("。") + } + limit := len(rep.Impacts) + if limit > 8 { + limit = 8 + } + for i := 0; i < limit; i++ { + imp := rep.Impacts[i] + label := imp.ItemKey + if label == "" { + label = imp.DocID + } + if label == "" { + continue + } + b.WriteString("\n- ") + b.WriteString(label) + if imp.Reason != "" { + b.WriteString("：") + b.WriteString(imp.Reason) + } + } + if len(rep.Impacts) > limit { + fmt.Fprintf(&b, "\n另有 %d 条未展开。", len(rep.Impacts)-limit) + } + return b.String() } func projectIDFromControlledDoc(docID string) int64 { @@ -86,6 +127,42 @@ func projectIDFromControlledDoc(docID string) int64 { } return n } + +func bindRecheckChat(svc *traceimpact.Service, chat *ctxmgr.ContextManager) { + if svc == nil || chat == nil { + return + } + svc.SetRecheckHook(func(ctx context.Context, mark traceimpact.RecheckMark) { + userID := traceimpact.ActorID(ctx) + reason := mark.Reason + if reason == "" { + reason = mark.SourceDoc + } + content := fmt.Sprintf("阶段 %s 待复核（%s）。需要细节可以说「分析影响」。", mark.StageKey, reason) + card := map[string]any{ + "cardId": fmt.Sprintf("recheck-%d", mark.ID), + "cardType": "RECHECK_NOTICE", + "title": "阶段待复核", + "proposedContent": content, + "status": "active", + "ttlTurns": 8, + "actions": []any{ + map[string]any{ + "actionId": "recheck_no_change", + "label": "无需变更", + "style": "primary", + "payload": map[string]string{ + "project_id": strconv.FormatInt(mark.ProjectID, 10), + "mark_id": strconv.FormatInt(mark.ID, 10), + }, + }, + }, + } + _ = docnotify.PostProjectNotice(ctx, chat, mark.ProjectID, userID, content, map[string]any{ + "interactive_card": card, + }) + }) +} func (t traceChat) ProposeRevision(ctx context.Context, projectID, userID int64, text string, confirmed bool) (string, error) { if t.svc == nil { return "", fmt.Errorf("trace: unavailable") diff --git a/internal/ctxmgr/message_update.go b/internal/ctxmgr/message_update.go index e4e655b0..c57a03c8 100644 --- a/internal/ctxmgr/message_update.go +++ b/internal/ctxmgr/message_update.go @@ -108,6 +108,48 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa return errors.New("message not found") } +// SetInteractiveCardStatus marks a stored chat card resolved or otherwise closed. +func (m *ContextManager) SetInteractiveCardStatus(ctx context.Context, messageID int64, status, summary string) error { + if m == nil || messageID <= 0 || status == "" { + return errors.New("invalid card status update") + } + if m.pool != nil || hasUoW(ctx) { + _, err := execChat(ctx, m.pool, ` + UPDATE chat_messages + SET metadata = jsonb_set( + jsonb_set(COALESCE(metadata, '{}'::jsonb), '{interactive_card,status}', to_jsonb($2::text), true), + '{interactive_card,resolutionSummary}', to_jsonb($3::text), true) + WHERE id = $1 + `, messageID, status, summary) + if err != nil { + return fmt.Errorf("set interactive card status: %w", err) + } + return nil + } + m.mu.Lock() + defer m.mu.Unlock() + for sid, msgs := range m.messages { + for i := range msgs { + if msgs[i].ID != messageID { + continue + } + card, _ := msgs[i].Metadata["interactive_card"].(map[string]any) + if card == nil { + card = map[string]any{} + } + card["status"] = status + card["resolutionSummary"] = summary + if msgs[i].Metadata == nil { + msgs[i].Metadata = map[string]any{} + } + msgs[i].Metadata["interactive_card"] = card + m.messages[sid][i] = msgs[i] + return nil + } + } + return errors.New("message not found") +} + func hasUoW(ctx context.Context) bool { _, ok := uow.ScopeFrom(ctx) return ok diff --git a/internal/intent/recognizer.go b/internal/intent/recognizer.go index 76ecb93a..d2c1ba33 100644 --- a/internal/intent/recognizer.go +++ b/internal/intent/recognizer.go @@ -1240,7 +1240,8 @@ func isGreetingOrHelp(input string) bool { func matchImpactIntent(input string) *CandidateIntent { propose := strings.Contains(input, "修订提案") || strings.Contains(input, "生成修订") || strings.Contains(input, "提议修改文档") - analyze := strings.Contains(input, "分析影响") || strings.Contains(input, "影响分析") || strings.Contains(input, "波及哪些") + analyze := strings.Contains(input, "分析影响") || strings.Contains(input, "影响分析") || strings.Contains(input, "波及哪些") || + strings.Contains(input, "影响哪些") || strings.Contains(input, "改动影响") || strings.Contains(input, "变更影响") if !propose && !analyze { return nil } @@ -1249,8 +1250,8 @@ func matchImpactIntent(input string) *CandidateIntent { intent = IntentRevisePropose } return &CandidateIntent{ - Intent: intent, - Confidence: 0.9, + Intent: intent, + Confidence: 0.9, FilledSlots: map[string]string{}, } } diff --git a/internal/intent/recognizer_test.go b/internal/intent/recognizer_test.go index ef747b2e..e0f4e2f6 100644 --- a/internal/intent/recognizer_test.go +++ b/internal/intent/recognizer_test.go @@ -773,3 +773,16 @@ func TestIntentRecognition_PermissionMismatch_NotHijackedBySupplement(t *testing t.Errorf("expected branch NEW_INTENT_PREEMPT for permission mismatch, got %s", res.Branch) } } + +func TestIntentRecognition_ImpactPhrases(t *testing.T) { + rec := intent.NewRecognizer() + for _, text := range []string{"请分析影响", "这次改动影响哪些文档", "帮我做一下变更影响"} { + res, err := rec.Recognize(context.Background(), text, &intent.ContextSnapshot{SnapshotID: "impact"}) + if err != nil { + t.Fatal(err) + } + if res.TopIntent.Intent != intent.IntentImpactAnalyze { + t.Fatalf("%q intent=%s", text, res.TopIntent.Intent) + } + } +} diff --git a/internal/server/trace_handlers.go b/internal/server/trace_handlers.go index 43fdb2e1..ecda0b2e 100644 --- a/internal/server/trace_handlers.go +++ b/internal/server/trace_handlers.go @@ -60,7 +60,7 @@ func (s *Server) handleProjectTraceIndex(w http.ResponseWriter, r *http.Request) } func (s *Server) handleProjectTraceAnalyze(w http.ResponseWriter, r *http.Request) { - projectID, _, ok := s.traceProject(w, r) + projectID, user, ok := s.traceProject(w, r) if !ok { return } @@ -72,7 +72,7 @@ func (s *Server) handleProjectTraceAnalyze(w http.ResponseWriter, r *http.Reques writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) return } - rep, err := s.rt.Trace.Analyze(r.Context(), projectID, body.DocRef, body.Others) + rep, err := s.rt.Trace.Analyze(traceimpact.WithActor(r.Context(), user.ID), projectID, body.DocRef, body.Others) if err != nil { writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) return @@ -176,7 +176,8 @@ func (s *Server) handleProjectTraceRecheckClear(w http.ResponseWriter, r *http.R return } var body struct { - Reason string `json:"reason"` + Reason string `json:"reason"` + MessageID int64 `json:"message_id"` } if err := json.NewDecoder(r.Body).Decode(&body); err != nil && err != io.EOF { writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) @@ -202,6 +203,9 @@ func (s *Server) handleProjectTraceRecheckClear(w http.ResponseWriter, r *http.R writeJSONResponse(w, http.StatusBadRequest, map[string]string{"error": "TRACE_RECHECK_CLEAR_FAILED", "message": err.Error()}) return } + if body.MessageID > 0 && s.ctxMgr != nil { + _ = s.ctxMgr.SetInteractiveCardStatus(r.Context(), body.MessageID, "resolved", "已标记为无需变更") + } writeJSONResponse(w, http.StatusOK, map[string]string{"status": "cleared"}) } diff --git a/internal/traceimpact/service.go b/internal/traceimpact/service.go index 43844786..eee97e04 100644 --- a/internal/traceimpact/service.go +++ b/internal/traceimpact/service.go @@ -21,9 +21,10 @@ type DocUnlocker interface { // Service is the project trace index and the stage-backtrack ledger. type Service struct { - store Store - reopen StageReopener - unlock DocUnlocker + store Store + reopen StageReopener + unlock DocUnlocker + onRecheck func(ctx context.Context, mark RecheckMark) } func NewService(store Store) *Service { @@ -40,6 +41,30 @@ func NewMemoryService() *Service { func (s *Service) SetStageReopener(r StageReopener) { s.reopen = r } func (s *Service) SetDocUnlocker(u DocUnlocker) { s.unlock = u } +// SetRecheckHook is called once when a new open recheck mark is stored. +// Repeat analyzes of the same source do not call it again. +func (s *Service) SetRecheckHook(fn func(ctx context.Context, mark RecheckMark)) { + if s != nil { + s.onRecheck = fn + } +} + +type actorCtxKey struct{} + +// WithActor records who triggered this analysis so a recheck notice can reach their chat. +func WithActor(ctx context.Context, userID int64) context.Context { + if userID <= 0 { + return ctx + } + return context.WithValue(ctx, actorCtxKey{}, userID) +} + +// ActorID is the user from WithActor, or 0. +func ActorID(ctx context.Context) int64 { + id, _ := ctx.Value(actorCtxKey{}).(int64) + return id +} + // IndexDocument projects explicit keys into items and same-line pairs into confirmed edges. // Untyped IDs become candidate items only. func (s *Service) IndexDocument(ctx context.Context, projectID int64, doc DocRef) (IndexState, error) { @@ -186,10 +211,22 @@ func (s *Service) Analyze(ctx context.Context, projectID int64, source DocRef, o Reason: fmt.Sprintf("%s 经已确认关系受本次修改影响", it.BusinessKey), }) if target.StageStatus == "completed" && target.StageKey != "" { - _, _ = s.store.PutMark(ctx, RecheckMark{ + already := false + if existing, listErr := s.store.ListOpenMarks(ctx, projectID, target.StageKey); listErr == nil { + for _, m := range existing { + if m.SourceDoc == source.DocID { + already = true + break + } + } + } + saved, putErr := s.store.PutMark(ctx, RecheckMark{ ProjectID: projectID, StageKey: target.StageKey, SourceDoc: source.DocID, Status: MarkOpen, Reason: "上游或同项目条目已变化", }) + if putErr == nil && !already && s.onRecheck != nil { + s.onRecheck(ctx, saved) + } } } } diff --git a/internal/traceimpact/service_test.go b/internal/traceimpact/service_test.go index 1847a4a3..3157dd35 100644 --- a/internal/traceimpact/service_test.go +++ b/internal/traceimpact/service_test.go @@ -65,6 +65,13 @@ func TestAnalyze_PendingIsNotNoImpact(t *testing.T) { func TestAnalyze_ConfirmedDownstreamAndRecheck(t *testing.T) { svc := NewMemoryService() ctx := t.Context() + var notices int + svc.SetRecheckHook(func(_ context.Context, mark RecheckMark) { + notices++ + if mark.StageKey != "V0.5" || mark.ID <= 0 { + t.Fatalf("mark %+v", mark) + } + }) if _, err := svc.IndexDocument(ctx, 7, DocRef{DocID: "up", Markdown: "UR-001 用户需要记录"}); err != nil { t.Fatal(err) } @@ -98,6 +105,17 @@ func TestAnalyze_ConfirmedDownstreamAndRecheck(t *testing.T) { if err := svc.ClearRecheck(ctx, marks[0].ID, "auto"); err == nil { t.Fatal("auto clear must fail") } + if notices != 1 { + t.Fatalf("recheck hook calls=%d", notices) + } + if _, err := svc.Analyze(ctx, 7, DocRef{DocID: "up", Markdown: "UR-001 改为必须记录"}, []DocRef{{ + DocID: "down", StageKey: "V0.5", StageStatus: "completed", Visible: true, + }}); err != nil { + t.Fatal(err) + } + if notices != 1 { + t.Fatalf("repeat analyze must not notify again, calls=%d", notices) + } if err := svc.ClearRecheck(ctx, marks[0].ID, ClearNoChange); err != nil { t.Fatal(err) }