diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index 048d6d6..0104371 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -371,6 +371,57 @@ func TestHITLChoiceFlow(t *testing.T) { } } +func TestConversationResumeSeq(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + _, convID := setupConversation(t, h) + runID := postMessage(t, h, convID, "#hitl") + + deadline := time.Now().Add(5 * time.Second) + var waiting map[string]any + for time.Now().Before(deadline) && waiting == nil { + code, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil) + if code == 200 { + if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 { + waiting = d + } + } + time.Sleep(30 * time.Millisecond) + } + if waiting == nil { + t.Fatal("no active hitl card") + } + lastSeq := int64(waiting["lastSeq"].(float64)) + resumeSeq := int64(waiting["resumeSeq"].(float64)) + var first int64 + for _, e := range replayEvents(t, st, convID) { + if string(e.RunID) == runID { + first = e.Seq + break + } + } + if first == 0 { + t.Fatal("run has no events") + } + if resumeSeq != first-1 { + t.Fatalf("resumeSeq = %d, want first-1 = %d", resumeSeq, first-1) + } + if resumeSeq >= lastSeq { + t.Fatalf("resumeSeq %d should be before lastSeq %d while the run is open", resumeSeq, lastSeq) + } + + cardID := waiting["activeHitl"].([]any)[0].(map[string]any)["id"].(string) + mustCode(t, h, "POST", "/api/hitl/"+cardID+"/resolve", map[string]any{"choiceId": "md"}) + terminal := waitRunTerminal(t, h, runID, 5*time.Second) + if terminal["status"] != "completed" { + t.Fatalf("run status after resolve: %v", terminal["status"]) + } + _, settled := doJSON(t, h, "GET", "/api/conversations/"+convID, nil) + if int64(settled["resumeSeq"].(float64)) != int64(settled["lastSeq"].(float64)) { + t.Fatalf("resumeSeq = %v, lastSeq = %v, want them equal after the run ends", settled["resumeSeq"], settled["lastSeq"]) + } +} + func TestHITLFreeTextPivot(t *testing.T) { srv, _ := newTestServer(t) h := srv.Handler() diff --git a/internal/server/server.go b/internal/server/server.go index 1164c01..c7b965a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -492,9 +492,16 @@ func (s *Server) conversationDetail(w http.ResponseWriter, r *http.Request) { runs, _ := s.st.ListRunsByConversation(id) cards, _ := s.st.ActiveHITLCards(owner, id) lastSeq, _ := s.st.LastEventSeq(id) + // resumeSeq:进行中的 run 从它的第一条事件之前重放,刷新或切回时补回已产生的过程。 + resumeSeq := lastSeq + if open, err := s.st.LatestOpenRun(id); err == nil { + if seq, ok, ferr := s.st.FirstRunEventSeq(id, open.ID); ferr == nil && ok && seq > 0 { + resumeSeq = seq - 1 + } + } writeJSON(w, http.StatusOK, map[string]any{ "conversation": conv, "messages": msgs, "runs": runs, - "activeHitl": cards, "lastSeq": lastSeq, + "activeHitl": cards, "lastSeq": lastSeq, "resumeSeq": resumeSeq, }) } diff --git a/internal/store/messages.go b/internal/store/messages.go index 9349ed8..8aaecec 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -338,6 +338,17 @@ func (s *Store) LastEventSeq(conversationID string) (int64, error) { return seq.Int64, nil } +// FirstRunEventSeq 返回这个 run 在会话事件流里的第一条 seq。 +// 没有事件时 ok 为 false。调用方用 seq-1 作为重放游标。 +func (s *Store) FirstRunEventSeq(conversationID, runID string) (seq int64, ok bool, err error) { + var got sql.NullInt64 + err = s.db.QueryRow(s.q(`SELECT MIN(seq) FROM events WHERE conversation_id=? AND run_id=?`), conversationID, runID).Scan(&got) + if err != nil || !got.Valid { + return 0, false, err + } + return got.Int64, true, nil +} + func newID(kind string) string { b := make([]byte, 6) rand.Read(b)