weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(trace): deliver recheck and impact notices in chat

Browse Source 
A new recheck mark is posted into the actor's existing project session, and a downstream-impact event does the same. Asking about what a change affects still resolves to impact analysis.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-21 05:19:43 +00:00
co-authored by Cursor
parent 2167bf4327
commit 687f930f40
13 changed files with 320 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/events/main.go
+2
View File
@@ -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,
internal/app/app.go
+2 -1
View File
@@ -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)
}
internal/app/docnotify/chat_notice.go
+45
View File
@@ -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 "有文档变更，下游内容可能受影响。需要时可以说「分析影响」。"
}
}
internal/app/docnotify/chat_notice_test.go
+51
View File
@@ -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)
}
}
internal/app/docnotify/consumer.go
+10 -1
View File
@@ -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 {
internal/app/docnotify/subscriber.go
+6 -1
View File
@@ -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
internal/app/trace_bridge.go
+79 -2
View File
@@ -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")
internal/ctxmgr/message_update.go
+42
View File
@@ -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
internal/intent/recognizer.go
+4 -3
View File
@@ -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{},
}
}
internal/intent/recognizer_test.go
+13
View File
@@ -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)
}
}
}
internal/server/trace_handlers.go
+7 -3
View File
@@ -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"})
}
internal/traceimpact/service.go
+41 -4
View File
@@ -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)
}
}
}
}
internal/traceimpact/service_test.go
+18
View File
@@ -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)
}