fix(store): session Save() UPSERT fallback, memory index-all user_id header (#379, #517)

- sessions_list.go (PG): check rowsAffected after UPDATE, INSERT with ON CONFLICT DO UPDATE
  when session not yet in DB (cron/heartbeat sessions)
- sessions_ops.go (SQLite): same UPSERT pattern with ? placeholders
- memory_handlers.go: fallback to X-GoClaw-User-Id header when body.user_id is empty
This commit is contained in:
viettranx committed 2026-03-30 14:12:08 +07:00
1 parent ec060ad7e1
commit 8eb4ce6d6f
3 files changed
+70 -4

No files matched your search

+3
View File
@@ -149,6 +149,9 @@ func (h *MemoryHandler) handleIndexAll(w http.ResponseWriter, r *http.Request) {
UserID string `json:"user_id"`
}
json.NewDecoder(r.Body).Decode(&body)
if body.UserID == "" {
body.UserID = extractUserID(r)
}
if err := h.store.IndexAll(r.Context(), agentID, body.UserID); err != nil {
slog.Warn("memory.index_all failed", "error", err)
+33 -2
View File
@@ -281,7 +281,7 @@ func (s *PGSessionStore) Save(ctx context.Context, key string) error {
metaJSON, _ = json.Marshal(snapshot.Metadata)
}
_, err := s.db.ExecContext(ctx,
res, err := s.db.ExecContext(ctx,
`UPDATE sessions SET
messages = $1, summary = $2, model = $3, provider = $4, channel = $5,
input_tokens = $6, output_tokens = $7, compaction_count = $8,
@@ -298,7 +298,38 @@ func (s *PGSessionStore) Save(ctx context.Context, key string) error {
snapshot.TeamID,
key, tenantIDForInsert(ctx),
)
return err
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
// Session not yet in DB (e.g. cron/heartbeat sessions) — insert it.
_, err = s.db.ExecContext(ctx,
`INSERT INTO sessions (id, session_key, messages, summary, model, provider, channel,
input_tokens, output_tokens, compaction_count,
memory_flush_compaction_count, memory_flush_at,
label, spawned_by, spawn_depth, agent_id, user_id, metadata, updated_at, team_id, tenant_id, created_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)
ON CONFLICT (tenant_id, session_key) DO UPDATE SET
messages = EXCLUDED.messages, summary = EXCLUDED.summary, model = EXCLUDED.model,
provider = EXCLUDED.provider, channel = EXCLUDED.channel,
input_tokens = EXCLUDED.input_tokens, output_tokens = EXCLUDED.output_tokens,
compaction_count = EXCLUDED.compaction_count,
memory_flush_compaction_count = EXCLUDED.memory_flush_compaction_count,
memory_flush_at = EXCLUDED.memory_flush_at,
label = EXCLUDED.label, spawned_by = EXCLUDED.spawned_by, spawn_depth = EXCLUDED.spawn_depth,
agent_id = EXCLUDED.agent_id, user_id = EXCLUDED.user_id, metadata = EXCLUDED.metadata,
updated_at = EXCLUDED.updated_at, team_id = EXCLUDED.team_id`,
uuid.Must(uuid.NewV7()), key, msgsJSON,
nilStr(snapshot.Summary), nilStr(snapshot.Model), nilStr(snapshot.Provider), nilStr(snapshot.Channel),
snapshot.InputTokens, snapshot.OutputTokens, snapshot.CompactionCount,
snapshot.MemoryFlushCompactionCount, snapshot.MemoryFlushAt,
nilStr(snapshot.Label), nilStr(snapshot.SpawnedBy), snapshot.SpawnDepth,
nilSessionUUID(snapshot.AgentUUID), nilStr(snapshot.UserID), metaJSON, snapshot.Updated,
snapshot.TeamID, tenantIDForInsert(ctx), snapshot.Updated,
)
return err
}
return nil
}
func (s *PGSessionStore) LastUsedChannel(ctx context.Context, agentID string) (string, string) {
+34 -2
View File
@@ -9,6 +9,7 @@ import (
"strings"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
@@ -31,7 +32,7 @@ func (s *SQLiteSessionStore) Save(ctx context.Context, key string) error {
metaJSON, _ = json.Marshal(snapshot.Metadata)
}
_, err := s.db.ExecContext(ctx,
res, err := s.db.ExecContext(ctx,
`UPDATE sessions SET
messages = ?, summary = ?, model = ?, provider = ?, channel = ?,
input_tokens = ?, output_tokens = ?, compaction_count = ?,
@@ -48,7 +49,38 @@ func (s *SQLiteSessionStore) Save(ctx context.Context, key string) error {
snapshot.TeamID,
key, tenantIDForInsert(ctx),
)
return err
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
// Session not yet in DB (e.g. cron/heartbeat sessions) — insert it.
_, err = s.db.ExecContext(ctx,
`INSERT INTO sessions (id, session_key, messages, summary, model, provider, channel,
input_tokens, output_tokens, compaction_count,
memory_flush_compaction_count, memory_flush_at,
label, spawned_by, spawn_depth, agent_id, user_id, metadata, updated_at, team_id, tenant_id, created_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(session_key, tenant_id) DO UPDATE SET
messages = excluded.messages, summary = excluded.summary, model = excluded.model,
provider = excluded.provider, channel = excluded.channel,
input_tokens = excluded.input_tokens, output_tokens = excluded.output_tokens,
compaction_count = excluded.compaction_count,
memory_flush_compaction_count = excluded.memory_flush_compaction_count,
memory_flush_at = excluded.memory_flush_at,
label = excluded.label, spawned_by = excluded.spawned_by, spawn_depth = excluded.spawn_depth,
agent_id = excluded.agent_id, user_id = excluded.user_id, metadata = excluded.metadata,
updated_at = excluded.updated_at, team_id = excluded.team_id`,
uuid.Must(uuid.NewV7()), key, msgsJSON,
nilStr(snapshot.Summary), nilStr(snapshot.Model), nilStr(snapshot.Provider), nilStr(snapshot.Channel),
snapshot.InputTokens, snapshot.OutputTokens, snapshot.CompactionCount,
snapshot.MemoryFlushCompactionCount, snapshot.MemoryFlushAt,
nilStr(snapshot.Label), nilStr(snapshot.SpawnedBy), snapshot.SpawnDepth,
nilSessionUUID(snapshot.AgentUUID), nilStr(snapshot.UserID), metaJSON, snapshot.Updated,
snapshot.TeamID, tenantIDForInsert(ctx), snapshot.Updated,
)
return err
}
return nil
}
func (s *SQLiteSessionStore) TruncateHistory(ctx context.Context, key string, keepLast int) {