mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
fix(serve): a stale title result must not overwrite a newer one, and a changed first message is a new request
This commit is contained in:
1 parent
d53132fdf6
commit
df88dde9ba
3 files changed
+83
-15
No files matched your search
@@ -149,3 +149,40 @@ func TestPreviewTitleStripsOnlyLeadingPasteLabel(t *testing.T) {
|
||||
t.Fatalf("inline label changed to %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// gatedTitleProvider holds its first request until released, so a test can ask
|
||||
// for a second title while the first generation is still running.
|
||||
type gatedTitleProvider struct {
|
||||
recordingTitleProvider
|
||||
entered chan struct{}
|
||||
release chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (p *gatedTitleProvider) Stream(ctx context.Context, req provider.Request) (<-chan provider.Chunk, error) {
|
||||
p.once.Do(func() {
|
||||
close(p.entered)
|
||||
<-p.release
|
||||
})
|
||||
return p.recordingTitleProvider.Stream(ctx, req)
|
||||
}
|
||||
|
||||
// A request for a different first message must not be taken for the one already
|
||||
// being generated: dropping it leaves the session showing a stale title until
|
||||
// some later list happens to ask again.
|
||||
func TestSessionTitleForNewSourceIsNotDroppedWhileTheOldOneGenerates(t *testing.T) {
|
||||
prov := &gatedTitleProvider{entered: make(chan struct{}), release: make(chan struct{})}
|
||||
s := &Server{titleModel: titleProviderState{prov: prov}, titles: newTitleCache(testenv.TempDir(t)), fill: newTitleFiller()}
|
||||
|
||||
s.sessionTitle("a.jsonl", "first prompt", 100, "")
|
||||
<-prov.entered
|
||||
if got := s.sessionTitle("a.jsonl", "replacement prompt", 300, ""); got != "replacement prompt" {
|
||||
t.Fatalf("preview = %q", got)
|
||||
}
|
||||
close(prov.release)
|
||||
|
||||
waitTitle(t, s, "a.jsonl", "replacement prompt", 300)
|
||||
if prov.count() != 2 {
|
||||
t.Fatalf("generations = %d, want 2: the replacement was dropped as a duplicate of the first", prov.count())
|
||||
}
|
||||
}
|
||||
@@ -99,6 +99,13 @@ func (c *titleCache) put(name, title, source string, mod int64) {
|
||||
_ = c.store(name, titleEntry{Title: title, Mod: mod, SourceHash: titleSourceHash(source)})
|
||||
}
|
||||
|
||||
// putIf stores the title only if keep still says it is wanted, asked while the
|
||||
// cache is locked so a stale result cannot land between the question and the write.
|
||||
func (c *titleCache) putIf(name, title, source string, mod int64, keep func() bool) bool {
|
||||
stored, err := c.storeIf(name, titleEntry{Title: title, Mod: mod, SourceHash: titleSourceHash(source)}, keep)
|
||||
return stored && err == nil
|
||||
}
|
||||
|
||||
func (c *titleCache) putExplicit(name, title string) error {
|
||||
return c.store(name, titleEntry{Title: title, Explicit: true})
|
||||
}
|
||||
@@ -123,27 +130,36 @@ func (c *titleCache) clearExplicit(name string) error {
|
||||
}
|
||||
|
||||
func (c *titleCache) store(name string, entry titleEntry) error {
|
||||
_, err := c.storeIf(name, entry, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
// storeIf reports whether the entry was written; a nil keep always writes.
|
||||
func (c *titleCache) storeIf(name string, entry titleEntry, keep func() bool) (bool, error) {
|
||||
titleCacheMu.Lock()
|
||||
defer titleCacheMu.Unlock()
|
||||
entries, err := c.load()
|
||||
if err != nil {
|
||||
return err
|
||||
return false, err
|
||||
}
|
||||
if entries[name].Explicit && !entry.Explicit {
|
||||
return nil
|
||||
return false, nil
|
||||
}
|
||||
if keep != nil && !keep() {
|
||||
return false, nil
|
||||
}
|
||||
entries[name] = entry
|
||||
data, err := json.Marshal(entries)
|
||||
if err != nil {
|
||||
return err
|
||||
return false, err
|
||||
}
|
||||
writeCache := func() error {
|
||||
return fileutil.AtomicWriteFile(filepath.Join(c.dir, ".session-titles.json"), data, 0o644)
|
||||
}
|
||||
if entry.Explicit {
|
||||
return writeSessionTitle(filepath.Join(c.dir, name), entry.Title, writeCache)
|
||||
return true, writeSessionTitle(filepath.Join(c.dir, name), entry.Title, writeCache)
|
||||
}
|
||||
return writeCache()
|
||||
return true, writeCache()
|
||||
}
|
||||
|
||||
// Hold the metadata lock through both writes and rollback so a failed cache
|
||||
|
||||
@@ -30,27 +30,40 @@ type titleJob struct {
|
||||
type titleFiller struct {
|
||||
mu sync.Mutex
|
||||
pending map[string]struct{}
|
||||
queue []titleJob
|
||||
active int
|
||||
// latest is the first message most recently asked for per session: a result
|
||||
// for any other is stale and is not stored over a newer one.
|
||||
latest map[string]string
|
||||
queue []titleJob
|
||||
active int
|
||||
}
|
||||
|
||||
// wanted reports whether source is still the first message to title for name.
|
||||
func (f *titleFiller) wanted(name, source string) bool {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.latest[name] == source
|
||||
}
|
||||
|
||||
func newTitleFiller() *titleFiller {
|
||||
return &titleFiller{pending: map[string]struct{}{}}
|
||||
return &titleFiller{pending: map[string]struct{}{}, latest: map[string]string{}}
|
||||
}
|
||||
|
||||
// scheduleTitle queues one generation, deduplicated by session name so a burst
|
||||
// of list requests cannot stack N copies of the same call.
|
||||
// scheduleTitle queues one generation, deduplicated by session and first message
|
||||
// so a burst of list requests cannot stack N copies of the same call, while a
|
||||
// changed first message is a new request and is never taken for the old one.
|
||||
func (s *Server) scheduleTitle(name, source string, mod int64) {
|
||||
if nilutil.IsNil(s.currentTitleProvider().prov) || source == "" {
|
||||
return
|
||||
}
|
||||
f := s.fill
|
||||
key := name + "\x00" + source
|
||||
f.mu.Lock()
|
||||
if _, dup := f.pending[name]; dup {
|
||||
f.latest[name] = source
|
||||
if _, dup := f.pending[key]; dup {
|
||||
f.mu.Unlock()
|
||||
return
|
||||
}
|
||||
f.pending[name] = struct{}{}
|
||||
f.pending[key] = struct{}{}
|
||||
f.queue = append(f.queue, titleJob{name: name, source: source, mod: mod})
|
||||
spawn := f.active < titleFillWorkers
|
||||
if spawn {
|
||||
@@ -78,13 +91,15 @@ func (s *Server) drainTitles() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), titleFillTimeout)
|
||||
title := s.generateTitle(ctx, job.source)
|
||||
cancel()
|
||||
if title != "" {
|
||||
s.titles.put(job.name, title, job.source, job.mod)
|
||||
if title != "" && s.titles.putIf(job.name, title, job.source, job.mod, func() bool { return f.wanted(job.name, job.source) }) {
|
||||
s.bc.treeChanged()
|
||||
}
|
||||
|
||||
f.mu.Lock()
|
||||
delete(f.pending, job.name)
|
||||
delete(f.pending, job.name+"\x00"+job.source)
|
||||
if f.latest[job.name] == job.source {
|
||||
delete(f.latest, job.name)
|
||||
}
|
||||
f.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user