opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(pkg): session, skills, state, runtime updates

Browse source
This commit is contained in:
ZanzyTHEbar 2026-02-22 22:49:38 +00:00
parent 189fc943e4
commit c2719ff30f
5 changed files with 97 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
pkg/runtime/config.go
View file

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

46
pkg/session/manager.go
View file

@ -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

7
pkg/session/manager_test.go
View file

@ -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"))

20
pkg/skills/loader.go
View file

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

35
pkg/state/state.go
View file

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