diff --git a/cmd/mavend/voice.go b/cmd/mavend/voice.go index a72281e..e5af719 100644 --- a/cmd/mavend/voice.go +++ b/cmd/mavend/voice.go @@ -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() diff --git a/internal/dialogue/session.go b/internal/dialogue/session.go index a573c99..d7203f3 100644 --- a/internal/dialogue/session.go +++ b/internal/dialogue/session.go @@ -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 { diff --git a/internal/dialogue/session_persist_test.go b/internal/dialogue/session_persist_test.go new file mode 100644 index 0000000..fc4f308 --- /dev/null +++ b/internal/dialogue/session_persist_test.go @@ -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) + } +} diff --git a/internal/store/dialogue.go b/internal/store/dialogue.go new file mode 100644 index 0000000..2d58db3 --- /dev/null +++ b/internal/store/dialogue.go @@ -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 +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 2c34c80..2bb3b52 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -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