From 60f37adc04cb8f93b73d7b86964198e4f692d5f6 Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Tue, 20 Oct 2026 03:51:12 +0000 Subject: [PATCH] agent-exec transport 9b5caca2968f510a4f81402ca7dfdb4c/f6d16260d971 --- internal/runmgr/recovery_test.go | 119 +++++++++++ internal/store/content.go | 24 ++- internal/store/messages.go | 9 +- internal/store/recovery.go | 128 +++++++++--- internal/store/run_account_write.go | 10 +- .../store/search_deadlock_run_timing_test.go | 189 ++++++++++++++++++ internal/store/search_postgres.go | 22 +- internal/store/vectors.go | 27 ++- 8 files changed, 477 insertions(+), 51 deletions(-) create mode 100644 internal/store/search_deadlock_run_timing_test.go diff --git a/internal/runmgr/recovery_test.go b/internal/runmgr/recovery_test.go index 7712254..1b7947c 100644 --- a/internal/runmgr/recovery_test.go +++ b/internal/runmgr/recovery_test.go @@ -147,6 +147,28 @@ func TestReconcileInterruptedSealsOpenRuns(t *testing.T) { if failed != 3 { t.Fatalf("run.failed events: %d", failed) } + msgs, err := st.ListMessages(conv.ID) + if err != nil { + t.Fatal(err) + } + asstByRun := map[string]int{} + for _, m := range msgs { + if m.Role == core.RoleAssistant { + asstByRun[m.RunID]++ + if len(m.Parts) != 1 || m.Parts[0].Type != core.PartError || m.Parts[0].Err == nil || m.Parts[0].Err.Code != "run.interrupted" || m.Parts[0].Err.Message != "进程重启或停机，运行已中断" { + t.Fatalf("unexpected assistant message parts for run %s: %+v", m.RunID, m.Parts) + } + } + } + for _, id := range []string{pending.ID, running.ID, waiting.ID} { + if asstByRun[id] != 1 { + t.Fatalf("run %s: expected 1 assistant message, got %d", id, asstByRun[id]) + } + } + if asstByRun[done.ID] != 0 { + t.Fatalf("terminal run %s: expected 0 assistant messages, got %d", done.ID, asstByRun[done.ID]) + } + if err := mgr.ReconcileInterrupted(); err != nil { t.Fatal(err) } @@ -157,6 +179,103 @@ func TestReconcileInterruptedSealsOpenRuns(t *testing.T) { if len(again) != len(evs) { t.Fatalf("second reconcile emitted more events: %d -> %d", len(evs), len(again)) } + msgsAgain, err := st.ListMessages(conv.ID) + if err != nil { + t.Fatal(err) + } + if len(msgsAgain) != len(msgs) { + t.Fatalf("second reconcile added duplicate messages: %d -> %d", len(msgs), len(msgsAgain)) + } +} + +func TestReconcileInterruptedAssistantNotice(t *testing.T) { + st := openRecoveryStore(t) + st.SetInstanceID("this") + mgr := newRecoveryManager(t, st) + mgr.SetInstanceID("this") + + // 1. Conversation with an open run that gets sealed + conv1 := recoveryConversation(t, st) + userMsg1 := core.Message{ConversationID: conv1.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hello"}}} + if err := st.AppendMessage(&userMsg1); err != nil { + t.Fatal(err) + } + sealedRun := recoveryRun(t, st, conv1, core.RunRunning) + + // 2. Conversation with an already terminal run + conv2 := recoveryConversation(t, st) + userMsg2 := core.Message{ConversationID: conv2.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}} + if err := st.AppendMessage(&userMsg2); err != nil { + t.Fatal(err) + } + terminalRun := recoveryRun(t, st, conv2, core.RunCompleted) + + // 3. Conversation with an open run that already has an assistant message + conv3 := recoveryConversation(t, st) + userMsg3 := core.Message{ConversationID: conv3.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "test"}}} + if err := st.AppendMessage(&userMsg3); err != nil { + t.Fatal(err) + } + openRunWithMsg := recoveryRun(t, st, conv3, core.RunRunning) + asstMsg3 := core.Message{ConversationID: conv3.ID, RunID: openRunWithMsg.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "already answered"}}} + if err := st.AppendMessage(&asstMsg3); err != nil { + t.Fatal(err) + } + + if err := mgr.ReconcileInterrupted(); err != nil { + t.Fatal(err) + } + + // conv1: history contains exactly one assistant message with run.interrupted + msgs1, err := st.ListMessages(conv1.ID) + if err != nil { + t.Fatal(err) + } + var asst1 []core.Message + for _, m := range msgs1 { + if m.Role == core.RoleAssistant { + asst1 = append(asst1, m) + } + } + if len(asst1) != 1 { + t.Fatalf("conv1: expected exactly 1 assistant message, got %d", len(asst1)) + } + if asst1[0].RunID != sealedRun.ID { + t.Fatalf("conv1: expected assistant message for run %s, got %s", sealedRun.ID, asst1[0].RunID) + } + if len(asst1[0].Parts) != 1 || asst1[0].Parts[0].Type != core.PartError || asst1[0].Parts[0].Err == nil || asst1[0].Parts[0].Err.Code != "run.interrupted" || asst1[0].Parts[0].Err.Message != "进程重启或停机，运行已中断" { + t.Fatalf("conv1: unexpected parts: %+v", asst1[0].Parts) + } + + // conv2: already-terminal run must have NO assistant message written + kept, err := st.GetRunByID(terminalRun.ID) + if err != nil || kept.Status != core.RunCompleted { + t.Fatalf("terminal run changed: %+v %v", kept, err) + } + msgs2, err := st.ListMessages(conv2.ID) + if err != nil { + t.Fatal(err) + } + for _, m := range msgs2 { + if m.Role == core.RoleAssistant { + t.Fatalf("conv2: unexpected assistant message for already-terminal run: %+v", m) + } + } + + // conv3: run had an assistant message already, so no duplicate was created + msgs3, err := st.ListMessages(conv3.ID) + if err != nil { + t.Fatal(err) + } + var asst3 []core.Message + for _, m := range msgs3 { + if m.Role == core.RoleAssistant { + asst3 = append(asst3, m) + } + } + if len(asst3) != 1 || asst3[0].ID != asstMsg3.ID { + t.Fatalf("conv3: expected only original assistant message, got %+v", asst3) + } } func TestReconcileLeavesOtherInstanceOnPostgres(t *testing.T) { diff --git a/internal/store/content.go b/internal/store/content.go index c63d31a..5a9464e 100644 --- a/internal/store/content.go +++ b/internal/store/content.go @@ -805,12 +805,20 @@ func (s *Store) searchKnowledgeFTSCandidates(pred knowledgePred, q core.Retrieva return nil, err } defer rows.Close() - out := []core.RetrievalHit{} + var rawHits []core.RetrievalHit for rows.Next() { h, err := scanKnowledgeHit(rows) if err != nil { return nil, err } + rawHits = append(rawHits, h) + } + if err := rows.Err(); err != nil { + return nil, err + } + _ = rows.Close() + out := []core.RetrievalHit{} + for _, h := range rawHits { if !pred(h.Doc) { continue } @@ -819,7 +827,7 @@ func (s *Store) searchKnowledgeFTSCandidates(pred knowledgePred, q core.Retrieva break } } - return out, rows.Err() + return out, nil } func (s *Store) searchKnowledgeLikeShared(requester string, q core.RetrievalQuery, pred knowledgePred) ([]core.RetrievalHit, error) { @@ -842,7 +850,7 @@ func (s *Store) searchKnowledgeLikeSharedCandidates(requester string, q core.Ret return nil, err } defer rows.Close() - out := []core.RetrievalHit{} + var rawDocs []core.KnowledgeDoc for rows.Next() { var d core.KnowledgeDoc var own, ca string @@ -851,6 +859,14 @@ func (s *Store) searchKnowledgeLikeSharedCandidates(requester string, q core.Ret } d.Owner = own d.CreatedAt, _ = parseTime(ca) + rawDocs = append(rawDocs, d) + } + if err := rows.Err(); err != nil { + return nil, err + } + _ = rows.Close() + out := []core.RetrievalHit{} + for _, d := range rawDocs { if !pred(d) { continue } @@ -859,7 +875,7 @@ func (s *Store) searchKnowledgeLikeSharedCandidates(requester string, q core.Ret break } } - return out, rows.Err() + return out, nil } func (s *Store) searchKnowledgeLike(owner string, q core.RetrievalQuery, topK int) ([]core.RetrievalHit, error) { diff --git a/internal/store/messages.go b/internal/store/messages.go index b450f1c..1800203 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -210,9 +210,14 @@ func (s *Store) CreateRun(r *core.Run) error { if r.ID == "" { r.ID = newID("run") } - r.CreatedAt = r.CreatedAt.UTC() + if r.CreatedAt.IsZero() { + r.CreatedAt = time.Now().UTC() + } else { + r.CreatedAt = r.CreatedAt.UTC() + } + ca := r.CreatedAt.Format(time.RFC3339Nano) _, err := s.db.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,instance_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`), - r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO(), s.instanceID) + r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, ca, s.instanceID) return err } diff --git a/internal/store/recovery.go b/internal/store/recovery.go index 3674046..71b77a7 100644 --- a/internal/store/recovery.go +++ b/internal/store/recovery.go @@ -86,15 +86,70 @@ func (s *Store) ListRecoverableRuns(instanceID string, all bool) ([]core.Run, er // SealInterrupted 把仍在进行的 run 收成 interrupted。 // 条件更新保证并发对账时只有一个赢家；已经是终态的行不受影响。 +// 赢家并在同一事务中为该 run 写入一条助手中断错误消息（若尚无助手消息）。 func (s *Store) SealInterrupted(id string) (bool, error) { - errPart := marshalJSON(&core.ErrorPart{Code: "run.interrupted", Message: "进程重启或停机，运行已中断"}) - res, err := s.db.Exec(s.q(`UPDATE runs SET status=?, phase=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`), - string(core.RunInterrupted), string(core.PhaseDone), errPart, nowISO(), id) + tx, err := s.db.Begin() + if err != nil { + return false, err + } + defer tx.Rollback() + + var convID string + err = tx.QueryRow(s.q(`SELECT conversation_id FROM runs WHERE id=?`), id).Scan(&convID) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } else if err != nil { + return false, err + } + if convID != "" { + if err := s.dialect.lockConversation(tx, convID); err != nil { + return false, err + } + } + + errPartObj := &core.ErrorPart{Code: "run.interrupted", Message: "进程重启或停机，运行已中断"} + errPartJSON := marshalJSON(errPartObj) + res, err := tx.Exec(s.q(`UPDATE runs SET status=?, phase=?, error=?, ended_at=? WHERE id=? AND status IN ('pending','running','waiting_hitl')`), + string(core.RunInterrupted), string(core.PhaseDone), errPartJSON, nowISO(), id) if err != nil { return false, err } n, err := res.RowsAffected() - return n == 1, err + if err != nil { + return false, err + } + if n != 1 { + return false, nil + } + + var count int + if err := tx.QueryRow(s.q(`SELECT COUNT(*) FROM messages WHERE run_id=? AND role=?`), id, string(core.RoleAssistant)).Scan(&count); err != nil { + return false, err + } + if count == 0 && convID != "" { + msg := &core.Message{ + ID: newID("msg"), + ConversationID: convID, + RunID: id, + Role: core.RoleAssistant, + Parts: []core.Part{{Type: core.PartError, Err: errPartObj}}, + CreatedAt: time.Now().UTC(), + } + var variantGroup string + var variantIndex int + if err := tx.QueryRow(s.q(`SELECT variant_group, variant_index FROM regen_pending WHERE conversation_id=?`), convID).Scan(&variantGroup, &variantIndex); err == nil && variantGroup != "" { + msg.VariantGroup = variantGroup + msg.VariantIndex = variantIndex + } + if err := s.insertSettledMessageTx(tx, msg); err != nil { + return false, err + } + } + + if err := tx.Commit(); err != nil { + return false, err + } + return true, nil } // ActiveHITLCardsByRun 返回某个 run 上尚未关闭的 HITL 卡片。 @@ -256,32 +311,7 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef, a return false, false, err } if msg != nil { - if msg.CreatedAt.IsZero() { - msg.CreatedAt = time.Now().UTC() - } - insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) - SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` - if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), msg.CreatedAt.UTC().Format(time.RFC3339Nano), "", msg.ConversationID); err != nil { - return false, false, err - } - if err := s.appendProjectMessageObservationsTx(tx, msg); err != nil { - return false, false, err - } - if msg.VariantGroup != "" || msg.Excluded { - ex := 0 - if msg.Excluded { - ex = 1 - } - if _, err := tx.Exec(s.q(`UPDATE messages SET excluded=?, variant_group=?, variant_index=? WHERE id=?`), ex, msg.VariantGroup, msg.VariantIndex, msg.ID); err != nil { - return false, false, err - } - } - if msg.VariantGroup != "" { - if _, err := tx.Exec(s.q(`DELETE FROM regen_pending WHERE conversation_id=? AND variant_group=? AND variant_index=?`), msg.ConversationID, msg.VariantGroup, msg.VariantIndex); err != nil { - return false, false, err - } - } - if _, err := tx.Exec(s.q(`UPDATE conversations SET updated_at=? WHERE id=?`), nowISO(), msg.ConversationID); err != nil { + if err := s.insertSettledMessageTx(tx, msg); err != nil { return false, false, err } } @@ -291,6 +321,44 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef, a return n == 1, false, nil } +func (s *Store) insertSettledMessageTx(tx *sql.Tx, msg *core.Message) error { + if msg == nil { + return nil + } + if msg.ID == "" { + msg.ID = newID("msg") + } + if msg.CreatedAt.IsZero() { + msg.CreatedAt = time.Now().UTC() + } + insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) + SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` + if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), msg.CreatedAt.UTC().Format(time.RFC3339Nano), "", msg.ConversationID); err != nil { + return err + } + if err := s.appendProjectMessageObservationsTx(tx, msg); err != nil { + return err + } + if msg.VariantGroup != "" || msg.Excluded { + ex := 0 + if msg.Excluded { + ex = 1 + } + if _, err := tx.Exec(s.q(`UPDATE messages SET excluded=?, variant_group=?, variant_index=? WHERE id=?`), ex, msg.VariantGroup, msg.VariantIndex, msg.ID); err != nil { + return err + } + } + if msg.VariantGroup != "" { + if _, err := tx.Exec(s.q(`DELETE FROM regen_pending WHERE conversation_id=? AND variant_group=? AND variant_index=?`), msg.ConversationID, msg.VariantGroup, msg.VariantIndex); err != nil { + return err + } + } + if _, err := tx.Exec(s.q(`UPDATE conversations SET updated_at=? WHERE id=?`), nowISO(), msg.ConversationID); err != nil { + return err + } + return nil +} + // ListInterruptedSubagentRuns 列出被收成 interrupted 的子代理 run // （N2 WP5 重启续跑的候选集；幂等过滤由转录目录的 done/resumed 标记完成）。 func (s *Store) ListInterruptedSubagentRuns() ([]core.Run, error) { diff --git a/internal/store/run_account_write.go b/internal/store/run_account_write.go index 636b086..f5d57b6 100644 --- a/internal/store/run_account_write.go +++ b/internal/store/run_account_write.go @@ -3,6 +3,7 @@ package store import ( "database/sql" "errors" + "time" core "agent-system-core" ) @@ -42,7 +43,12 @@ func (s *Store) CreateRunAccountChecked(r *core.Run, allowImplicitDefault bool) if r.ID == "" { r.ID = newID("run") } - r.CreatedAt = r.CreatedAt.UTC() + if r.CreatedAt.IsZero() { + r.CreatedAt = time.Now().UTC() + } else { + r.CreatedAt = r.CreatedAt.UTC() + } + ca := r.CreatedAt.Format(time.RFC3339Nano) return s.withEffectiveAccountWrite(string(r.Owner), allowImplicitDefault, func(tx *sql.Tx, _ Account) error { var owner string if err := tx.QueryRow(s.q(`SELECT owner FROM conversations WHERE id=?`), r.ConversationID).Scan(&owner); errors.Is(err, sql.ErrNoRows) { @@ -54,7 +60,7 @@ func (s *Store) CreateRunAccountChecked(r *core.Run, allowImplicitDefault bool) return ErrNotFound } _, err := tx.Exec(s.q(`INSERT INTO runs(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,instance_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`), - r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, nowISO(), s.instanceID) + r.ID, r.ParentRunID, r.ConversationID, r.ProjectID, r.Owner, string(r.Kind), r.ProfileID, string(r.Status), string(r.Phase), r.AgentName, ca, s.instanceID) return err }) } diff --git a/internal/store/search_deadlock_run_timing_test.go b/internal/store/search_deadlock_run_timing_test.go new file mode 100644 index 0000000..d36e712 --- /dev/null +++ b/internal/store/search_deadlock_run_timing_test.go @@ -0,0 +1,189 @@ +package store + +import ( + "context" + "testing" + "time" + + core "agent-system-core" +) + +func TestSearchDeadlockRegression(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + + // Seed account for allow/pred callback to query + if err := st.CreateAccount("alice", "", "", false); err != nil { + t.Fatal(err) + } + + // Seed knowledge doc and vector + docVec := core.KnowledgeDoc{ + ID: "doc_vec_deadlock", + Owner: "alice", + ProjectID: "proj_deadlock", + Source: "src_vec", + Title: "vector doc", + Chunk: "vector chunk content", + } + docFTS := core.KnowledgeDoc{ + ID: "doc_fts_deadlock", + Owner: "alice", + ProjectID: "proj_deadlock", + Source: "src_fts", + Title: "Database Optimization Guide", + Chunk: "Performance tuning and indexing for relational databases", + } + docLike := core.KnowledgeDoc{ + ID: "doc_like_deadlock", + Owner: "alice", + ProjectID: "proj_deadlock", + Source: "src_like", + Title: "微服务架构治理", + Chunk: "分布式系统与全链路追踪方案", + } + + if err := st.AddKnowledge(docVec, docFTS, docLike); err != nil { + t.Fatal(err) + } + if err := st.UpsertKnowledgeVector(ctx, "alice", docVec.ID, "ref_vec", []float32{1.0, 0.0}); err != nil { + t.Fatal(err) + } + + runBounded := func(name string, fn func()) { + t.Helper() + done := make(chan struct{}) + go func() { + defer close(done) + fn() + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatalf("deadlock detected in %s: did not finish within 10s", name) + } + } + + // 1. Vector scan: allow callback queries the same Store + runBounded("searchKnowledgeVectorsSharedAuthorizedScan", func() { + hits, err := st.SearchKnowledgeVectorsSharedAuthorized(ctx, "bob", "", []float32{1.0, 0.0}, 5, core.RetrievalQuery{}, func(d core.KnowledgeDoc) bool { + // Query the store inside the callback: on a single-connection SQLite store, + // this would deadlock if rows was not closed before invoking the callback. + _, qerr := st.GetAccountByOwner(d.Owner) + if qerr != nil { + t.Errorf("GetAccountByOwner failed in callback: %v", qerr) + } + return true + }) + if err != nil { + t.Errorf("SearchKnowledgeVectorsSharedAuthorized failed: %v", err) + } + if len(hits) == 0 { + t.Errorf("expected at least 1 hit, got 0") + } + }) + + // 2. Content FTS scan: pred callback queries the same Store + runBounded("searchKnowledgeFTSCandidates", func() { + hits, err := st.searchKnowledgeFTSCandidates(func(d core.KnowledgeDoc) bool { + _, qerr := st.GetAccountByOwner(d.Owner) + if qerr != nil { + t.Errorf("GetAccountByOwner failed in callback: %v", qerr) + } + return true + }, core.RetrievalQuery{Text: "database optimization", TopK: 5}, true) + if err != nil { + t.Errorf("searchKnowledgeFTSCandidates failed: %v", err) + } + if len(hits) == 0 { + t.Errorf("expected at least 1 hit, got 0") + } + }) + + // 3. Content LIKE scan: pred callback queries the same Store + runBounded("searchKnowledgeLikeSharedCandidates", func() { + hits, err := st.searchKnowledgeLikeSharedCandidates("bob", core.RetrievalQuery{Text: "微服务", TopK: 5}, func(d core.KnowledgeDoc) bool { + _, qerr := st.GetAccountByOwner(d.Owner) + if qerr != nil { + t.Errorf("GetAccountByOwner failed in callback: %v", qerr) + } + return true + }, true) + if err != nil { + t.Errorf("searchKnowledgeLikeSharedCandidates failed: %v", err) + } + if len(hits) == 0 { + t.Errorf("expected at least 1 hit, got 0") + } + }) +} + +func TestRunCreatedAtNonZeroRecentAndMatchesReadBack(t *testing.T) { + st := openTestStore(t) + + if err := st.CreateAccount("alice", "", "", false); err != nil { + t.Fatal(err) + } + proj, err := st.CreateProject("alice", "test_proj") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation("alice", proj.ID, "test_conv", "") + if err != nil { + t.Fatal(err) + } + + // 1. CreateRun with zero CreatedAt + before1 := time.Now().UTC().Add(-time.Second) + r1 := core.Run{ + ConversationID: conv.ID, + ProjectID: conv.ProjectID, + Owner: conv.Owner, + Kind: core.RunConversation, + Status: core.RunRunning, + Phase: core.PhaseThinking, + } + if err := st.CreateRun(&r1); err != nil { + t.Fatalf("CreateRun failed: %v", err) + } + if r1.CreatedAt.IsZero() { + t.Fatalf("CreateRun: returned run has zero CreatedAt") + } + if r1.CreatedAt.Before(before1) || r1.CreatedAt.After(time.Now().UTC().Add(time.Second)) { + t.Fatalf("CreateRun: returned run CreatedAt %v not recent (before=%v)", r1.CreatedAt, before1) + } + readBack1, err := st.GetRunByID(r1.ID) + if err != nil { + t.Fatalf("GetRunByID failed: %v", err) + } + if !readBack1.CreatedAt.Equal(r1.CreatedAt) { + t.Fatalf("CreateRun: in-memory CreatedAt %v != stored CreatedAt %v", r1.CreatedAt, readBack1.CreatedAt) + } + + // 2. CreateRunAccountChecked with zero CreatedAt + before2 := time.Now().UTC().Add(-time.Second) + r2 := core.Run{ + ConversationID: conv.ID, + ProjectID: conv.ProjectID, + Owner: conv.Owner, + Kind: core.RunConversation, + Status: core.RunRunning, + Phase: core.PhaseThinking, + } + if err := st.CreateRunAccountChecked(&r2, false); err != nil { + t.Fatalf("CreateRunAccountChecked failed: %v", err) + } + if r2.CreatedAt.IsZero() { + t.Fatalf("CreateRunAccountChecked: returned run has zero CreatedAt") + } + if r2.CreatedAt.Before(before2) || r2.CreatedAt.After(time.Now().UTC().Add(time.Second)) { + t.Fatalf("CreateRunAccountChecked: returned run CreatedAt %v not recent (before=%v)", r2.CreatedAt, before2) + } + readBack2, err := st.GetRunByID(r2.ID) + if err != nil { + t.Fatalf("GetRunByID failed: %v", err) + } + if !readBack2.CreatedAt.Equal(r2.CreatedAt) { + t.Fatalf("CreateRunAccountChecked: in-memory CreatedAt %v != stored CreatedAt %v", r2.CreatedAt, readBack2.CreatedAt) + } +} diff --git a/internal/store/search_postgres.go b/internal/store/search_postgres.go index f5ffea5..e812ec6 100644 --- a/internal/store/search_postgres.go +++ b/internal/store/search_postgres.go @@ -79,7 +79,11 @@ func (s *Store) searchKnowledgePGSharedCandidates(requester string, q core.Retri return nil, err } defer rows.Close() - out := []core.RetrievalHit{} + type rawPGCandidate struct { + doc core.KnowledgeDoc + rank float64 + } + var rawCands []rawPGCandidate for rows.Next() { var d core.KnowledgeDoc var own, ca string @@ -89,17 +93,25 @@ func (s *Store) searchKnowledgePGSharedCandidates(requester string, q core.Retri } d.Owner = own d.CreatedAt, _ = parseTime(ca) - if !pred(d) { + rawCands = append(rawCands, rawPGCandidate{doc: d, rank: rank}) + } + if err := rows.Err(); err != nil { + return nil, err + } + _ = rows.Close() + out := []core.RetrievalHit{} + for _, c := range rawCands { + if !pred(c.doc) { continue } - score := rank + score := c.rank if score == 0 { score = 0.5 } - out = append(out, core.RetrievalHit{Doc: d, Score: score, Kind: "fulltext"}) + out = append(out, core.RetrievalHit{Doc: c.doc, Score: score, Kind: "fulltext"}) if len(out) >= topK { break } } - return out, rows.Err() + return out, nil } diff --git a/internal/store/vectors.go b/internal/store/vectors.go index f559698..ded73c0 100644 --- a/internal/store/vectors.go +++ b/internal/store/vectors.go @@ -279,11 +279,11 @@ func (s *Store) searchKnowledgeVectorsSharedAuthorizedScan(ctx context.Context, return nil, err } defer rows.Close() - type cand struct { - hit core.RetrievalHit - sim float64 + type rawVectorCandidate struct { + doc core.KnowledgeDoc + blob []byte } - var cands []cand + var rawCands []rawVectorCandidate for rows.Next() { var d core.KnowledgeDoc var own, ca string @@ -293,6 +293,20 @@ func (s *Store) searchKnowledgeVectorsSharedAuthorizedScan(ctx context.Context, } d.Owner = own d.CreatedAt, _ = parseTime(ca) + rawCands = append(rawCands, rawVectorCandidate{doc: d, blob: blob}) + } + if err := rows.Err(); err != nil { + return nil, err + } + _ = rows.Close() + + type cand struct { + hit core.RetrievalHit + sim float64 + } + var cands []cand + for _, raw := range rawCands { + d := raw.doc if d.ProjectID == "" || !allow(d) { continue } @@ -302,12 +316,9 @@ func (s *Store) searchKnowledgeVectorsSharedAuthorizedScan(ctx context.Context, if len(filter.Sources) > 0 && !containsStr(filter.Sources, d.Source) { continue } - sim := cosine(vec, decodeVector(blob)) + sim := cosine(vec, decodeVector(raw.blob)) cands = append(cands, cand{hit: core.RetrievalHit{Doc: d, Score: sim, Kind: "vector"}, sim: sim}) } - if err := rows.Err(); err != nil { - return nil, err - } sort.SliceStable(cands, func(i, j int) bool { return cands[i].sim > cands[j].sim }) out := []core.RetrievalHit{} for _, c := range cands {