diff --git a/pkg/runtime/config.go b/pkg/runtime/config.go index 5091aa290..a89a5afde 100644 --- a/pkg/runtime/config.go +++ b/pkg/runtime/config.go @@ -49,11 +49,6 @@ func ResolveBaseConfigPath() string { } } - // Fall back to the default devcontainer host mount location. - if path := checkHostConfig("/host_home"); path != "" { - return path - } - // Prefer XDG standard path (~/.config/dragonscale/config.json) when present. xdgPath, err := config.DefaultConfigPath() if err == nil { diff --git a/pkg/session/manager.go b/pkg/session/manager.go index 5fc16ebdc..9b20ae977 100644 --- a/pkg/session/manager.go +++ b/pkg/session/manager.go @@ -50,6 +50,12 @@ func WithSessionDelegate(del memory.MemoryDelegate, agentID string) SessionOptio } } +type msgPersistItem struct { + sessionKey string + msg messages.Message + barrier chan struct{} // non-nil for flush barriers; worker closes it after draining prior items +} + type SessionManager struct { sessions map[string]*Session // primary store (always authoritative) lru *cache.LRU[string, bool] // tracks access order; value is just a presence flag @@ -58,6 +64,8 @@ type SessionManager struct { cfg SessionManagerConfig delegate memory.MemoryDelegate agentID string + msgChan chan msgPersistItem // async message persistence; nil when delegate is nil + msgDone chan struct{} // closed when the persist worker exits } func NewSessionManager(storage string, opts ...SessionOption) *SessionManager { @@ -88,6 +96,9 @@ func NewSessionManagerWithConfig(storage string, cfg SessionManagerConfig, opts if sm.delegate != nil { sm.loadSessionsFromDelegate() + sm.msgChan = make(chan msgPersistItem, 256) + sm.msgDone = make(chan struct{}) + go sm.msgPersistWorker() } else if storage != "" { os.MkdirAll(storage, 0755) sm.loadSessions() @@ -349,12 +360,24 @@ func (sm *SessionManager) AddFullMessage(sessionKey string, msg messages.Message session.Updated = time.Now() sm.touchLRU(sessionKey) - if sm.delegate != nil { - sm.persistMessageToDelegate(sessionKey, msg) + if sm.msgChan != nil { + sm.msgChan <- msgPersistItem{sessionKey: sessionKey, msg: msg} } } +// msgPersistWorker drains msgChan and writes messages to the delegate in the background. +func (sm *SessionManager) msgPersistWorker() { + defer close(sm.msgDone) + for item := range sm.msgChan { + if item.barrier != nil { + close(item.barrier) + continue + } + sm.persistMessageToDelegate(item.sessionKey, item.msg) + } +} + func (sm *SessionManager) persistMessageToDelegate(sessionKey string, msg messages.Message) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() @@ -530,6 +553,25 @@ func (sm *SessionManager) CleanupStale(maxAge time.Duration) int { func sanitizeFilename(key string) string { return strings.ReplaceAll(key, ":", "_") } + +// Flush blocks until all pending async message persists complete. +func (sm *SessionManager) Flush() { + if sm.msgChan == nil { + return + } + barrier := make(chan struct{}) + sm.msgChan <- msgPersistItem{barrier: barrier} + <-barrier +} + +// Close drains the async message persistence channel and waits for completion. +func (sm *SessionManager) Close() { + if sm.msgChan != nil { + close(sm.msgChan) + <-sm.msgDone + } +} + func (sm *SessionManager) Save(key string) error { if sm.delegate != nil { return nil diff --git a/pkg/session/manager_test.go b/pkg/session/manager_test.go index 45cf092f3..bd6a09b9a 100644 --- a/pkg/session/manager_test.go +++ b/pkg/session/manager_test.go @@ -224,6 +224,8 @@ func TestSessionManager_DelegatePersistence(t *testing.T) { t.Fatalf("expected 2 messages in-memory, got %d", len(history)) } + sm.Flush() + items, err := del.ListRecallItems(t.Context(), "test-agent", key, 100, 0) if err != nil { t.Fatalf("ListRecallItems: %v", err) @@ -285,6 +287,7 @@ func TestSessionManager_DelegateBootstrapPaginationAndOrder(t *testing.T) { for i := 0; i < totalMsgs; i++ { writer.AddMessage(sessionKey, "user", fmt.Sprintf("m-%04d", i)) } + writer.Flush() // New manager instance exercises delegate bootstrap restore path. reader := NewSessionManager("", WithSessionDelegate(del, "test-agent")) @@ -313,6 +316,7 @@ func TestSessionManager_ProjectionPointerPersistedAndRestored(t *testing.T) { sm.AddMessage(sessionKey, "user", "first") sm.AddMessage(sessionKey, "assistant", "second") sm.AddMessage(sessionKey, "user", "third") + sm.Flush() // Read persisted pointer from KV raw, err := del.GetKV(t.Context(), "test-agent", projectionPointerKey(sessionKey)) @@ -350,6 +354,7 @@ func TestSessionManager_ProjectionPointerUpdatedOnAppend(t *testing.T) { for i := 0; i < 4; i++ { sm.AddMessage(sessionKey, "user", fmt.Sprintf("msg-%d", i)) + sm.Flush() raw, err := del.GetKV(t.Context(), "test-agent", projectionPointerKey(sessionKey)) require.NoError(t, err) require.NotEmpty(t, raw) @@ -370,6 +375,7 @@ func TestSessionManager_IntegrityMismatchRestoreStillSucceeds(t *testing.T) { sm := NewSessionManager("", WithSessionDelegate(del, "test-agent")) sm.AddMessage(sessionKey, "user", "a") sm.AddMessage(sessionKey, "assistant", "b") + sm.Flush() // Corrupt the stored pointer to simulate prior state mismatch corrupt := ProjectionPointer{Count: 0} @@ -403,6 +409,7 @@ func TestSessionManager_ProjectionBackfillStatusPersistedOnBootstrap(t *testing. writer.AddMessage("session-a", "user", "hello") writer.AddMessage("session-a", "assistant", "hi") writer.AddMessage("session-b", "user", "task") + writer.Flush() _ = NewSessionManager("", WithSessionDelegate(del, "test-agent")) diff --git a/pkg/skills/loader.go b/pkg/skills/loader.go index 82add1982..7d8ba7856 100644 --- a/pkg/skills/loader.go +++ b/pkg/skills/loader.go @@ -8,6 +8,7 @@ import ( "path/filepath" "regexp" "strings" + "time" jsonv2 "github.com/go-json-experiment/json" ) @@ -76,6 +77,25 @@ func NewSkillsLoader(primarySkillsDir string, globalSkills string, builtinSkills } } +// DirsMtime returns the latest modification time across all skill directories. +// Used for cache invalidation — if no directories exist, returns zero time. +func (sl *SkillsLoader) DirsMtime() time.Time { + var latest time.Time + for _, dir := range []string{sl.primarySkills, sl.globalSkills, sl.builtinSkills} { + if dir == "" { + continue + } + info, err := os.Stat(dir) + if err != nil { + continue + } + if mt := info.ModTime(); mt.After(latest) { + latest = mt + } + } + return latest +} + func (sl *SkillsLoader) ListSkills() []SkillInfo { skills := make([]SkillInfo, 0) diff --git a/pkg/state/state.go b/pkg/state/state.go index 155fe1e2c..238f1e693 100644 --- a/pkg/state/state.go +++ b/pkg/state/state.go @@ -91,6 +91,15 @@ func NewManager(workspace string, opts ...Option) *Manager { func (sm *Manager) loadFromDelegate() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() + + // Prefer single-key format (state:json) + if v, err := sm.delegate.GetKV(ctx, kvAgentID, "state:json"); err == nil && v != "" { + if err := jsonv2.Unmarshal([]byte(v), sm.state); err == nil { + return + } + } + + // Fallback: legacy 3-key format if v, err := sm.delegate.GetKV(ctx, kvAgentID, "state:last_channel"); err == nil && v != "" { sm.state.LastChannel = v } @@ -126,6 +135,18 @@ func (sm *Manager) SetLastChatID(ctx context.Context, chatID string) error { return sm.persist(ctx) } +// SetChannelAndChatID atomically updates both channel and chat ID in a single persist. +func (sm *Manager) SetChannelAndChatID(ctx context.Context, channel, chatID string) error { + sm.mu.Lock() + defer sm.mu.Unlock() + + sm.state.LastChannel = channel + sm.state.LastChatID = chatID + sm.state.Timestamp = time.Now() + + return sm.persist(ctx) +} + // persist writes the current state to the delegate (KV) or file. // Must be called with the lock held. func (sm *Manager) persist(ctx context.Context) error { @@ -136,16 +157,12 @@ func (sm *Manager) persist(ctx context.Context) error { } func (sm *Manager) persistToDelegate(ctx context.Context) error { - ts := sm.state.Timestamp.Format(time.RFC3339Nano) - - if err := sm.delegate.UpsertKV(ctx, kvAgentID, "state:last_channel", sm.state.LastChannel); err != nil { - return fmt.Errorf("upsert last_channel: %w", err) + blob, err := jsonv2.Marshal(sm.state, jsontext.WithIndent("")) + if err != nil { + return fmt.Errorf("marshal state: %w", err) } - if err := sm.delegate.UpsertKV(ctx, kvAgentID, "state:last_chat_id", sm.state.LastChatID); err != nil { - return fmt.Errorf("upsert last_chat_id: %w", err) - } - if err := sm.delegate.UpsertKV(ctx, kvAgentID, "state:timestamp", ts); err != nil { - return fmt.Errorf("upsert timestamp: %w", err) + if err := sm.delegate.UpsertKV(ctx, kvAgentID, "state:json", string(blob)); err != nil { + return fmt.Errorf("upsert state: %w", err) } return nil }