Merge dialogue session persistence
This commit is contained in:
kami
+12
-1
@@ -227,7 +227,18 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
|
||||
}
|
||||
|
||||
// ----- dialogue (multi-turn slot carry-over; 2-min follow-up window) -----
|
||||
dialogueSessions := dialogue.NewSessionStore(2 * time.Minute)
|
||||
// Store-backed when the daemon passes a store, so a restart mid-conversation
|
||||
// keeps the thread (Vikunja #363). Sessions past their TTL are dropped on
|
||||
// load, never revived. Clarify's parked question stays in memory only.
|
||||
var dialogueSessions *dialogue.SessionStore
|
||||
if dataStore != nil {
|
||||
dialogueSessions = dialogue.NewPersistentSessionStore(2*time.Minute, dataStore)
|
||||
if err := dialogueSessions.Load(context.Background(), time.Now()); err != nil {
|
||||
log.Printf("dialogue: load saved sessions: %v", err)
|
||||
}
|
||||
} else {
|
||||
dialogueSessions = dialogue.NewSessionStore(2 * time.Minute)
|
||||
}
|
||||
clarifyStore := dialogue.NewClarifyStore(clarifyTTL)
|
||||
timeParser := router.NewPythonDateParser()
|
||||
|
||||
|
||||
@@ -1,8 +1,12 @@
|
||||
package dialogue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
type Intent string
|
||||
@@ -50,10 +54,22 @@ func (s *Session) IsExpired(now time.Time) bool {
|
||||
return now.After(s.Timestamp.Add(s.TTL))
|
||||
}
|
||||
|
||||
// SessionPersister — the bit of the store the session needs, as an interface
|
||||
// so tests can swap it out. Data is an opaque blob: the store never looks
|
||||
// inside, we encode the session as JSON here.
|
||||
type SessionPersister interface {
|
||||
SaveDialogueSession(ctx context.Context, id string, data []byte, ts time.Time, ttl time.Duration) error
|
||||
DeleteDialogueSession(ctx context.Context, id string) error
|
||||
LoadDialogueSessions(ctx context.Context, now time.Time) ([]store.DialogueSessionRow, error)
|
||||
}
|
||||
|
||||
// SessionStore keeps the live sessions in a map (the fast path) and mirrors
|
||||
// every write to the persister, so a daemon restart can load them back.
|
||||
type SessionStore struct {
|
||||
mu sync.RWMutex
|
||||
sessions map[string]*Session
|
||||
defaultTTL time.Duration
|
||||
persist SessionPersister // may be nil: memory only (tests, no-store paths)
|
||||
}
|
||||
|
||||
func NewSessionStore(defaultTTL time.Duration) *SessionStore {
|
||||
@@ -66,6 +82,43 @@ func NewSessionStore(defaultTTL time.Duration) *SessionStore {
|
||||
}
|
||||
}
|
||||
|
||||
// NewPersistentSessionStore — same store, but writes also go to the DB.
|
||||
// Call Load once after this to bring back sessions from a previous run.
|
||||
func NewPersistentSessionStore(defaultTTL time.Duration, p SessionPersister) *SessionStore {
|
||||
s := NewSessionStore(defaultTTL)
|
||||
s.persist = p
|
||||
return s
|
||||
}
|
||||
|
||||
// Load — read the saved sessions back into memory. Anything past its TTL is
|
||||
// dropped (and deleted from the DB by the store), never revived.
|
||||
func (s *SessionStore) Load(ctx context.Context, now time.Time) error {
|
||||
if s.persist == nil {
|
||||
return nil
|
||||
}
|
||||
rows, err := s.persist.LoadDialogueSessions(ctx, now)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for _, r := range rows {
|
||||
var sess Session
|
||||
if err := json.Unmarshal(r.Data, &sess); err != nil {
|
||||
// A blob we can't read is not worth failing a startup over.
|
||||
continue
|
||||
}
|
||||
if sess.TTL <= 0 {
|
||||
sess.TTL = r.TTL
|
||||
}
|
||||
if sess.IsExpired(now) {
|
||||
continue
|
||||
}
|
||||
s.sessions[r.ID] = &sess
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SessionStore) Get(id string, now time.Time) *Session {
|
||||
s.mu.RLock()
|
||||
sess, ok := s.sessions[id]
|
||||
@@ -87,12 +140,33 @@ func (s *SessionStore) Put(id string, sess *Session) {
|
||||
s.mu.Lock()
|
||||
s.sessions[id] = sess
|
||||
s.mu.Unlock()
|
||||
s.save(id, sess)
|
||||
}
|
||||
|
||||
func (s *SessionStore) Delete(id string) {
|
||||
s.mu.Lock()
|
||||
delete(s.sessions, id)
|
||||
s.mu.Unlock()
|
||||
if s.persist != nil {
|
||||
_ = s.persist.DeleteDialogueSession(context.Background(), id)
|
||||
}
|
||||
}
|
||||
|
||||
// save — mirror one session to the DB. Best effort: memory already has it, so
|
||||
// a write error costs us the restart safety net, not the current turn.
|
||||
func (s *SessionStore) save(id string, sess *Session) {
|
||||
if s.persist == nil {
|
||||
return
|
||||
}
|
||||
data, err := json.Marshal(sess)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
ts := sess.Timestamp
|
||||
if ts.IsZero() {
|
||||
ts = time.Now()
|
||||
}
|
||||
_ = s.persist.SaveDialogueSession(context.Background(), id, data, ts, sess.TTL)
|
||||
}
|
||||
|
||||
func InheritSlots(prev, cur Slots) Slots {
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
package dialogue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/store"
|
||||
)
|
||||
|
||||
// openStore — a store on disk, so a second handle can reopen the same file.
|
||||
func openStore(t *testing.T, path string) *store.Store {
|
||||
t.Helper()
|
||||
s, err := store.Open(context.Background(), path)
|
||||
if err != nil {
|
||||
t.Fatalf("open store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = s.Close() })
|
||||
return s
|
||||
}
|
||||
|
||||
// A session written before a restart comes back and still merges a follow-up.
|
||||
func TestSessionSurvivesRestart(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
path := filepath.Join(t.TempDir(), "maven_test.db")
|
||||
now := time.Now().UTC().Truncate(time.Millisecond)
|
||||
|
||||
first := openStore(t, path)
|
||||
before := NewPersistentSessionStore(2*time.Minute, first)
|
||||
before.Put("voice", &Session{
|
||||
Intent: IntentReminder,
|
||||
Slots: Slots{Text: "полить цветы", Time: now.Add(time.Hour), HasTime: true},
|
||||
Timestamp: now,
|
||||
TTL: 2 * time.Minute,
|
||||
})
|
||||
if err := first.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
// fresh handle, fresh in-memory map — as after a daemon restart
|
||||
after := openStore(t, path)
|
||||
reloaded := NewPersistentSessionStore(2*time.Minute, after)
|
||||
if err := reloaded.Load(ctx, now.Add(10*time.Second)); err != nil {
|
||||
t.Fatalf("Load: %v", err)
|
||||
}
|
||||
sess := reloaded.Get("voice", now.Add(10*time.Second))
|
||||
if sess == nil {
|
||||
t.Fatal("session did not survive the restart")
|
||||
}
|
||||
if sess.Intent != IntentReminder {
|
||||
t.Fatalf("intent = %q, want reminder", sess.Intent)
|
||||
}
|
||||
// the follow-up carries no text of its own; it must inherit the old one
|
||||
merged := InheritSlots(sess.Slots, Slots{Time: now.Add(2 * time.Hour), HasTime: true})
|
||||
if merged.Text != "полить цветы" {
|
||||
t.Fatalf("merged text = %q, want the earlier turn's text", merged.Text)
|
||||
}
|
||||
if !merged.Time.Equal(now.Add(2 * time.Hour)) {
|
||||
t.Fatalf("merged time = %v, want the follow-up's time", merged.Time)
|
||||
}
|
||||
}
|
||||
|
||||
// A session past its TTL is dead: a restart must not bring it back.
|
||||
func TestExpiredSessionNotResurrected(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
path := filepath.Join(t.TempDir(), "maven_test.db")
|
||||
now := time.Now().UTC().Truncate(time.Millisecond)
|
||||
|
||||
first := openStore(t, path)
|
||||
before := NewPersistentSessionStore(2*time.Minute, first)
|
||||
before.Put("voice", &Session{
|
||||
Intent: IntentReminder,
|
||||
Slots: Slots{Text: "полить цветы"},
|
||||
Timestamp: now,
|
||||
TTL: time.Minute,
|
||||
})
|
||||
if err := first.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
after := openStore(t, path)
|
||||
reloaded := NewPersistentSessionStore(2*time.Minute, after)
|
||||
later := now.Add(5 * time.Minute) // well past the 1-min TTL
|
||||
if err := reloaded.Load(ctx, later); err != nil {
|
||||
t.Fatalf("Load: %v", err)
|
||||
}
|
||||
if sess := reloaded.Get("voice", later); sess != nil {
|
||||
t.Fatalf("expired session came back: %+v", sess)
|
||||
}
|
||||
// and it is gone from the DB too, not just from memory
|
||||
rows, err := after.LoadDialogueSessions(ctx, later)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadDialogueSessions: %v", err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Fatalf("expired row still in the DB: %+v", rows)
|
||||
}
|
||||
}
|
||||
|
||||
// Delete removes the row as well, so an ended conversation stays ended.
|
||||
func TestDeleteRemovesPersistedSession(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
path := filepath.Join(t.TempDir(), "maven_test.db")
|
||||
now := time.Now().UTC().Truncate(time.Millisecond)
|
||||
|
||||
s := openStore(t, path)
|
||||
ss := NewPersistentSessionStore(2*time.Minute, s)
|
||||
ss.Put("voice", &Session{Intent: IntentChat, Timestamp: now, TTL: time.Minute})
|
||||
ss.Delete("voice")
|
||||
rows, err := s.LoadDialogueSessions(ctx, now)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadDialogueSessions: %v", err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Fatalf("row survived Delete: %+v", rows)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DialogueSessionRow — one saved follow-up session. Data is the session
|
||||
// encoded by the dialogue package; the store does not look inside it.
|
||||
type DialogueSessionRow struct {
|
||||
ID string
|
||||
Data []byte
|
||||
Ts time.Time
|
||||
TTL time.Duration
|
||||
Expires time.Time
|
||||
}
|
||||
|
||||
// SaveDialogueSession — write (or replace) the session for one dialogue id.
|
||||
// One row per id: a newer turn overwrites the older state.
|
||||
func (s *Store) SaveDialogueSession(ctx context.Context, id string, data []byte, ts time.Time, ttl time.Duration) error {
|
||||
expires := ts.Add(ttl)
|
||||
_, err := s.db.ExecContext(ctx, `
|
||||
INSERT INTO dialogue_sessions (id, data, ts, ttl_ms, expires_ts) VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET data = excluded.data,
|
||||
ts = excluded.ts,
|
||||
ttl_ms = excluded.ttl_ms,
|
||||
expires_ts = excluded.expires_ts`,
|
||||
id, data, ts.UnixMilli(), ttl.Milliseconds(), expires.UnixMilli())
|
||||
if err != nil {
|
||||
return fmt.Errorf("save dialogue session: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteDialogueSession — drop one session (ended, or expired).
|
||||
func (s *Store) DeleteDialogueSession(ctx context.Context, id string) error {
|
||||
if _, err := s.db.ExecContext(ctx, `DELETE FROM dialogue_sessions WHERE id = ?`, id); err != nil {
|
||||
return fmt.Errorf("delete dialogue session: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LoadDialogueSessions — return the sessions still alive at `now` and delete
|
||||
// the ones that already ran out. An expired session is dead: it never comes
|
||||
// back after a restart.
|
||||
func (s *Store) LoadDialogueSessions(ctx context.Context, now time.Time) ([]DialogueSessionRow, error) {
|
||||
if _, err := s.db.ExecContext(ctx,
|
||||
`DELETE FROM dialogue_sessions WHERE expires_ts <= ?`, now.UnixMilli()); err != nil {
|
||||
return nil, fmt.Errorf("prune dialogue sessions: %w", err)
|
||||
}
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
`SELECT id, data, ts, ttl_ms, expires_ts FROM dialogue_sessions ORDER BY id`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load dialogue sessions: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []DialogueSessionRow
|
||||
for rows.Next() {
|
||||
var r DialogueSessionRow
|
||||
var tsMilli, ttlMilli, expMilli int64
|
||||
if err := rows.Scan(&r.ID, &r.Data, &tsMilli, &ttlMilli, &expMilli); err != nil {
|
||||
return nil, fmt.Errorf("scan dialogue session: %w", err)
|
||||
}
|
||||
r.Ts = time.UnixMilli(tsMilli).UTC()
|
||||
r.TTL = time.Duration(ttlMilli) * time.Millisecond
|
||||
r.Expires = time.UnixMilli(expMilli).UTC()
|
||||
out = append(out, r)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("load dialogue sessions: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -74,6 +74,15 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
||||
`CREATE INDEX IF NOT EXISTS idx_nudges_snoozed ON nudges (outcome_ts) WHERE outcome = 'snoozed';`, // #8 — SnoozedUntil runs every tick; keep it off a full scan (Vikunja #364)
|
||||
`ALTER TABLE proposed_routines ADD COLUMN accepted_ts INTEGER;
|
||||
ALTER TABLE proposed_routines ADD COLUMN last_fired_ts INTEGER;`, // #9 — accepted routines keep firing (Vikunja #366): the tick loop needs to know when a routine was accepted and when it last nudged
|
||||
|
||||
`CREATE TABLE IF NOT EXISTS dialogue_sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
data BLOB NOT NULL,
|
||||
ts INTEGER NOT NULL,
|
||||
ttl_ms INTEGER NOT NULL,
|
||||
expires_ts INTEGER NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_dialogue_sessions_expires ON dialogue_sessions (expires_ts);`, // #10 — the follow-up session survives a restart (Vikunja #363); small, TTL-pruned table, not a history log
|
||||
}
|
||||
|
||||
// migrate applies every migration with a number greater than the DB's current
|
||||
|
||||
Reference in New Issue
Block a user