feat: IM式跨端实时同步——message_persisted 广播 + 前端按id去重追加
- history.Store 新增 AppendFull(返回含 id/createdAt 的完整消息),Append 签名不变 - session_orchestrator 加可空 OnPersisted;persistUserMessages/persistAssistantReply 落库后回调广播 - server.go newOrchestrator 接 OnPersisted → hub.broadcast message_persisted(thinking 剥除) - 前端 ws.ts 加 message_persisted case(仅 activeSessionId 匹配才处理);appendOrUpdateMessage 升级为 id 精确upsert+内容/时间归一 - 实测:双 WS,A stream 发消息,B 免刷新收到 user+assistant 各1条(count=2 PASS) - 红线: 身份仍靠 actor 元数据,不靠提示词;用户/助手消息、本端/异端统一路径
This commit is contained in:
1 parent
bcc53d68f3
commit
1bf60c1ae4
9 files changed
+260
-38
No files matched your search
@@ -157,11 +157,22 @@ type fakeHistory struct {
|
||||
appends []history.Message
|
||||
}
|
||||
|
||||
func (f *fakeHistory) Append(_ context.Context, msg history.Message) error {
|
||||
func (f *fakeHistory) Append(ctx context.Context, msg history.Message) error {
|
||||
_, err := f.AppendFull(ctx, msg)
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *fakeHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if msg.ID == "" {
|
||||
msg.ID = fmt.Sprintf("id-%d", len(f.appends)+1)
|
||||
}
|
||||
if msg.CreatedAt == 0 {
|
||||
msg.CreatedAt = int64(len(f.appends) + 1)
|
||||
}
|
||||
f.appends = append(f.appends, msg)
|
||||
f.mu.Unlock()
|
||||
return nil
|
||||
return msg, nil
|
||||
}
|
||||
|
||||
func (f *fakeHistory) List(context.Context, string, int64, int) ([]history.Message, error) {
|
||||
@@ -219,6 +230,44 @@ func TestSubmitPersistsUserAndAssistant(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubmitPersistedBroadcast(t *testing.T) {
|
||||
t.Parallel()
|
||||
run := newGatedRunner(1)
|
||||
hist := &fakeHistory{}
|
||||
var mu sync.Mutex
|
||||
var persisted []history.Message
|
||||
o := NewOrchestrator(OrchestratorOptions{
|
||||
Run: run.Run,
|
||||
Log: slog.Default(),
|
||||
History: hist,
|
||||
OnPersisted: func(m history.Message) {
|
||||
mu.Lock()
|
||||
persisted = append(persisted, m)
|
||||
mu.Unlock()
|
||||
},
|
||||
})
|
||||
run.tickets <- struct{}{}
|
||||
sr, _ := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv())
|
||||
if !sr.Accepted || !sr.Immediate {
|
||||
t.Fatalf("submit = %+v, want Accepted+Immediate", sr)
|
||||
}
|
||||
mu.Lock()
|
||||
got := append([]history.Message(nil), persisted...)
|
||||
mu.Unlock()
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("persisted = %d, want 2 (user + assistant)", len(got))
|
||||
}
|
||||
if got[0].Role != history.RoleUser || got[0].ID == "" || got[0].SessionID != "web" || got[0].Content != "你好" {
|
||||
t.Fatalf("user persisted = %+v", got[0])
|
||||
}
|
||||
if got[1].Role != history.RoleAssistant || got[1].ID == "" || got[1].SessionID != "web" || got[1].Content == "" {
|
||||
t.Fatalf("assistant persisted = %+v", got[1])
|
||||
}
|
||||
if got[0].ID == got[1].ID {
|
||||
t.Fatalf("user and assistant share id %q", got[0].ID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubmitDoesNotPersistRejectedTurn(t *testing.T) {
|
||||
t.Parallel()
|
||||
hist := &fakeHistory{}
|
||||
|
||||
Reference in new issue
Block a user