agent-exec transport 9b5caca2968f510a4f81402ca7dfdb4c/f6d16260d971

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) {

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) {

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
 }
 

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) {

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
 	})
 }

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)
+	}
+}

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
 }

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 {