feat(session): persist scope metadata and aliases

This commit is contained in:
Hoshina
2026-04-01 16:25:05 +08:00
parent bb2167e3f3
commit 3957e2cc72
7 changed files with 585 additions and 104 deletions
+41
View File
@@ -74,6 +74,7 @@ type AgentLoop struct {
// processOptions configures how a message is processed
type processOptions struct {
SessionKey string // Session identifier for history/context
SessionAliases []string // Compatibility aliases for the session key
Channel string // Target channel for tool execution
ChatID string // Target chat ID for tool execution
MessageID string // Current inbound platform message ID
@@ -1475,6 +1476,7 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
opts := processOptions{
SessionKey: sessionKey,
SessionAliases: buildSessionAliases(sessionKey, allocation.SessionKey, msg.SessionKey),
Channel: msg.Channel,
ChatID: msg.ChatID,
MessageID: msg.MessageID,
@@ -1547,6 +1549,43 @@ func resolveScopeKey(routeSessionKey, msgSessionKey string) string {
return routeSessionKey
}
func buildSessionAliases(canonicalKey string, keys ...string) []string {
if len(keys) == 0 {
return nil
}
aliases := make([]string, 0, len(keys))
seen := make(map[string]struct{}, len(keys))
canonicalKey = strings.TrimSpace(canonicalKey)
for _, key := range keys {
key = strings.TrimSpace(key)
if key == "" || key == canonicalKey {
continue
}
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
aliases = append(aliases, key)
}
if len(aliases) == 0 {
return nil
}
return aliases
}
func ensureSessionMetadata(store session.SessionStore, key string, scope *session.SessionScope, aliases []string) {
if key == "" || scope == nil {
return
}
metaStore, ok := store.(interface {
EnsureSessionMetadata(sessionKey string, scope *session.SessionScope, aliases []string)
})
if !ok {
return
}
metaStore.EnsureSessionMetadata(key, scope, aliases)
}
func (al *AgentLoop) allocateRouteSession(route routing.ResolvedRoute, msg bus.InboundMessage) session.Allocation {
return session.AllocateRouteSession(session.AllocationInput{
AgentID: route.AgentID,
@@ -1668,6 +1707,8 @@ func (al *AgentLoop) runAgentLoop(
}
}
ensureSessionMetadata(agent.Sessions, opts.SessionKey, opts.SessionScope, opts.SessionAliases)
turnScope := al.newTurnEventScope(
agent.ID,
opts.SessionKey,
+157 -14
View File
@@ -32,14 +32,19 @@ const (
maxLineSize = 10 * 1024 * 1024 // 10 MB
)
// sessionMeta holds per-session metadata stored in a .meta.json file.
type sessionMeta struct {
Key string `json:"key"`
Summary string `json:"summary"`
Skip int `json:"skip"`
Count int `json:"count"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// SessionMeta holds per-session metadata stored in a .meta.json file.
//
// Scope is stored as raw JSON so pkg/memory can stay decoupled from the
// higher-level session package while still preserving structured scope data.
type SessionMeta struct {
Key string `json:"key"`
Summary string `json:"summary"`
Skip int `json:"skip"`
Count int `json:"count"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
Scope json.RawMessage `json:"scope,omitempty"`
Aliases []string `json:"aliases,omitempty"`
}
// JSONLStore implements Store using append-only JSONL files.
@@ -98,25 +103,31 @@ func sanitizeKey(key string) string {
// readMeta loads the metadata file for a session.
// Returns a zero-value sessionMeta if the file does not exist.
func (s *JSONLStore) readMeta(key string) (sessionMeta, error) {
func (s *JSONLStore) readMeta(key string) (SessionMeta, error) {
data, err := os.ReadFile(s.metaPath(key))
if os.IsNotExist(err) {
return sessionMeta{Key: key}, nil
return SessionMeta{Key: key}, nil
}
if err != nil {
return sessionMeta{}, fmt.Errorf("memory: read meta: %w", err)
return SessionMeta{}, fmt.Errorf("memory: read meta: %w", err)
}
var meta sessionMeta
var meta SessionMeta
err = json.Unmarshal(data, &meta)
if err != nil {
return sessionMeta{}, fmt.Errorf("memory: decode meta: %w", err)
return SessionMeta{}, fmt.Errorf("memory: decode meta: %w", err)
}
if meta.Key == "" {
meta.Key = key
}
return meta, nil
}
// writeMeta atomically writes the metadata file using the project's
// standard WriteFileAtomic (temp + fsync + rename).
func (s *JSONLStore) writeMeta(key string, meta sessionMeta) error {
func (s *JSONLStore) writeMeta(key string, meta SessionMeta) error {
if strings.TrimSpace(meta.Key) == "" {
meta.Key = key
}
data, err := json.MarshalIndent(meta, "", " ")
if err != nil {
return fmt.Errorf("memory: encode meta: %w", err)
@@ -124,6 +135,138 @@ func (s *JSONLStore) writeMeta(key string, meta sessionMeta) error {
return fileutil.WriteFileAtomic(s.metaPath(key), data, 0o644)
}
func cloneRawJSON(data json.RawMessage) json.RawMessage {
if len(data) == 0 {
return nil
}
return append(json.RawMessage(nil), data...)
}
func normalizeAliases(canonicalKey string, aliases []string) []string {
if len(aliases) == 0 {
return nil
}
normalized := make([]string, 0, len(aliases))
seen := make(map[string]struct{}, len(aliases))
canonicalKey = strings.TrimSpace(canonicalKey)
for _, alias := range aliases {
alias = strings.TrimSpace(alias)
if alias == "" || alias == canonicalKey {
continue
}
if _, ok := seen[alias]; ok {
continue
}
seen[alias] = struct{}{}
normalized = append(normalized, alias)
}
if len(normalized) == 0 {
return nil
}
return normalized
}
func (s *JSONLStore) sessionExists(key string) bool {
if key == "" {
return false
}
if _, err := os.Stat(s.jsonlPath(key)); err == nil {
return true
}
if _, err := os.Stat(s.metaPath(key)); err == nil {
return true
}
return false
}
// GetSessionMeta returns the current metadata snapshot for sessionKey.
func (s *JSONLStore) GetSessionMeta(_ context.Context, sessionKey string) (SessionMeta, error) {
l := s.sessionLock(sessionKey)
l.Lock()
defer l.Unlock()
meta, err := s.readMeta(sessionKey)
if err != nil {
return SessionMeta{}, err
}
meta.Scope = cloneRawJSON(meta.Scope)
if len(meta.Aliases) > 0 {
meta.Aliases = append([]string(nil), meta.Aliases...)
}
return meta, nil
}
// UpsertSessionMeta stores structured session metadata while preserving
// summary/count/skip timestamps maintained by the core JSONL store.
func (s *JSONLStore) UpsertSessionMeta(
_ context.Context,
sessionKey string,
scope json.RawMessage,
aliases []string,
) error {
l := s.sessionLock(sessionKey)
l.Lock()
defer l.Unlock()
meta, err := s.readMeta(sessionKey)
if err != nil {
return err
}
meta.Scope = cloneRawJSON(scope)
meta.Aliases = normalizeAliases(sessionKey, aliases)
now := time.Now()
if meta.CreatedAt.IsZero() {
meta.CreatedAt = now
}
meta.UpdatedAt = now
return s.writeMeta(sessionKey, meta)
}
// ResolveSessionKey returns the canonical session key for a candidate key.
// It first checks direct key existence, then scans metadata aliases on miss.
func (s *JSONLStore) ResolveSessionKey(_ context.Context, sessionKey string) (string, bool, error) {
sessionKey = strings.TrimSpace(sessionKey)
if sessionKey == "" {
return "", false, nil
}
if s.sessionExists(sessionKey) {
return sessionKey, true, nil
}
entries, err := os.ReadDir(s.dir)
if err != nil {
return "", false, fmt.Errorf("memory: read sessions dir: %w", err)
}
for _, entry := range entries {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".meta.json") {
continue
}
data, readErr := os.ReadFile(filepath.Join(s.dir, entry.Name()))
if readErr != nil {
return "", false, fmt.Errorf("memory: read meta: %w", readErr)
}
var meta SessionMeta
if err := json.Unmarshal(data, &meta); err != nil {
return "", false, fmt.Errorf("memory: decode meta: %w", err)
}
if meta.Key == "" {
continue
}
if meta.Key == sessionKey {
return meta.Key, true, nil
}
for _, alias := range meta.Aliases {
if alias == sessionKey {
return meta.Key, true, nil
}
}
}
return "", false, nil
}
// readMessages reads valid JSON lines from a .jsonl file, skipping
// the first `skip` lines without unmarshaling them. This avoids the
// cost of json.Unmarshal on logically truncated messages.
+55
View File
@@ -2,8 +2,10 @@ package memory
import (
"context"
"encoding/json"
"os"
"path/filepath"
"reflect"
"sync"
"testing"
@@ -241,6 +243,59 @@ func TestSetSummary_GetSummary(t *testing.T) {
}
}
func TestSessionMetaScopeAndAliasesPersist(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
scope := json.RawMessage(`{"version":1,"channel":"telegram","values":{"chat":"group:c1"}}`)
aliases := []string{"legacy:one", "legacy:one", "canonical"}
if err := store.UpsertSessionMeta(ctx, "canonical", scope, aliases); err != nil {
t.Fatalf("UpsertSessionMeta() error = %v", err)
}
meta, err := store.GetSessionMeta(ctx, "canonical")
if err != nil {
t.Fatalf("GetSessionMeta() error = %v", err)
}
var gotScope map[string]any
if err := json.Unmarshal(meta.Scope, &gotScope); err != nil {
t.Fatalf("Unmarshal(meta.Scope) error = %v", err)
}
var wantScope map[string]any
if err := json.Unmarshal(scope, &wantScope); err != nil {
t.Fatalf("Unmarshal(scope) error = %v", err)
}
if !reflect.DeepEqual(gotScope, wantScope) {
t.Fatalf("meta.Scope = %#v, want %#v", gotScope, wantScope)
}
if len(meta.Aliases) != 1 || meta.Aliases[0] != "legacy:one" {
t.Fatalf("meta.Aliases = %#v, want [legacy:one]", meta.Aliases)
}
}
func TestResolveSessionKeyByAlias(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
if err := store.AddMessage(ctx, "canonical", "user", "hello"); err != nil {
t.Fatalf("AddMessage() error = %v", err)
}
if err := store.UpsertSessionMeta(ctx, "canonical", nil, []string{"legacy:key"}); err != nil {
t.Fatalf("UpsertSessionMeta() error = %v", err)
}
resolved, found, err := store.ResolveSessionKey(ctx, "legacy:key")
if err != nil {
t.Fatalf("ResolveSessionKey() error = %v", err)
}
if !found {
t.Fatal("ResolveSessionKey() did not find alias")
}
if resolved != "canonical" {
t.Fatalf("resolved = %q, want %q", resolved, "canonical")
}
}
func TestTruncateHistory_KeepLast(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
+64
View File
@@ -2,6 +2,7 @@ package session
import (
"context"
"encoding/json"
"log"
"github.com/sipeed/picoclaw/pkg/memory"
@@ -15,24 +16,82 @@ type JSONLBackend struct {
store memory.Store
}
type metaAwareStore interface {
GetSessionMeta(ctx context.Context, sessionKey string) (memory.SessionMeta, error)
UpsertSessionMeta(ctx context.Context, sessionKey string, scope json.RawMessage, aliases []string) error
ResolveSessionKey(ctx context.Context, sessionKey string) (string, bool, error)
}
// MetadataAwareSessionStore exposes structured session metadata operations.
type MetadataAwareSessionStore interface {
EnsureSessionMetadata(sessionKey string, scope *SessionScope, aliases []string)
ResolveSessionKey(sessionKey string) string
}
// NewJSONLBackend wraps a memory.Store for use as a SessionStore.
func NewJSONLBackend(store memory.Store) *JSONLBackend {
return &JSONLBackend{store: store}
}
func (b *JSONLBackend) resolveSessionKey(sessionKey string) string {
metaStore, ok := b.store.(metaAwareStore)
if !ok {
return sessionKey
}
resolved, found, err := metaStore.ResolveSessionKey(context.Background(), sessionKey)
if err != nil {
log.Printf("session: resolve session key: %v", err)
return sessionKey
}
if found && resolved != "" {
return resolved
}
return sessionKey
}
// ResolveSessionKey maps aliases onto their canonical session key when the
// underlying store supports structured metadata. Unknown aliases fall back to
// the original input so existing callers remain compatible.
func (b *JSONLBackend) ResolveSessionKey(sessionKey string) string {
return b.resolveSessionKey(sessionKey)
}
// EnsureSessionMetadata persists scope and alias metadata for a session.
func (b *JSONLBackend) EnsureSessionMetadata(sessionKey string, scope *SessionScope, aliases []string) {
metaStore, ok := b.store.(metaAwareStore)
if !ok {
return
}
var rawScope json.RawMessage
if scope != nil {
data, err := json.Marshal(scope)
if err != nil {
log.Printf("session: encode session scope: %v", err)
return
}
rawScope = data
}
if err := metaStore.UpsertSessionMeta(context.Background(), sessionKey, rawScope, aliases); err != nil {
log.Printf("session: upsert session metadata: %v", err)
}
}
func (b *JSONLBackend) AddMessage(sessionKey, role, content string) {
sessionKey = b.resolveSessionKey(sessionKey)
if err := b.store.AddMessage(context.Background(), sessionKey, role, content); err != nil {
log.Printf("session: add message: %v", err)
}
}
func (b *JSONLBackend) AddFullMessage(sessionKey string, msg providers.Message) {
sessionKey = b.resolveSessionKey(sessionKey)
if err := b.store.AddFullMessage(context.Background(), sessionKey, msg); err != nil {
log.Printf("session: add full message: %v", err)
}
}
func (b *JSONLBackend) GetHistory(key string) []providers.Message {
key = b.resolveSessionKey(key)
msgs, err := b.store.GetHistory(context.Background(), key)
if err != nil {
log.Printf("session: get history: %v", err)
@@ -42,6 +101,7 @@ func (b *JSONLBackend) GetHistory(key string) []providers.Message {
}
func (b *JSONLBackend) GetSummary(key string) string {
key = b.resolveSessionKey(key)
summary, err := b.store.GetSummary(context.Background(), key)
if err != nil {
log.Printf("session: get summary: %v", err)
@@ -51,18 +111,21 @@ func (b *JSONLBackend) GetSummary(key string) string {
}
func (b *JSONLBackend) SetSummary(key, summary string) {
key = b.resolveSessionKey(key)
if err := b.store.SetSummary(context.Background(), key, summary); err != nil {
log.Printf("session: set summary: %v", err)
}
}
func (b *JSONLBackend) SetHistory(key string, history []providers.Message) {
key = b.resolveSessionKey(key)
if err := b.store.SetHistory(context.Background(), key, history); err != nil {
log.Printf("session: set history: %v", err)
}
}
func (b *JSONLBackend) TruncateHistory(key string, keepLast int) {
key = b.resolveSessionKey(key)
if err := b.store.TruncateHistory(context.Background(), key, keepLast); err != nil {
log.Printf("session: truncate history: %v", err)
}
@@ -72,6 +135,7 @@ func (b *JSONLBackend) TruncateHistory(key string, keepLast int) {
// immediately, the data is already durable. Save runs compaction to reclaim
// space from logically truncated messages (no-op when there are none).
func (b *JSONLBackend) Save(key string) error {
key = b.resolveSessionKey(key)
return b.store.Compact(context.Background(), key)
}
+28
View File
@@ -177,3 +177,31 @@ func TestJSONLBackend_SummarizeFlow(t *testing.T) {
t.Errorf("first message = %q, want %q", history[0].Content, "msg 16")
}
}
func TestJSONLBackend_ResolveAliasAndPersistMetadata(t *testing.T) {
b := newBackend(t)
b.EnsureSessionMetadata("canonical", &session.SessionScope{
Version: session.ScopeVersionV1,
AgentID: "main",
Channel: "telegram",
Account: "default",
Dimensions: []string{"chat"},
Values: map[string]string{
"chat": "group:c1",
},
}, []string{"legacy"})
if got := b.ResolveSessionKey("legacy"); got != "canonical" {
t.Fatalf("ResolveSessionKey() = %q, want %q", got, "canonical")
}
b.AddMessage("legacy", "user", "hello through alias")
history := b.GetHistory("canonical")
if len(history) != 1 {
t.Fatalf("len(history) = %d, want 1", len(history))
}
if history[0].Content != "hello through alias" {
t.Fatalf("history[0].Content = %q, want %q", history[0].Content, "hello through alias")
}
}