diff --git a/internal/http/memory_handlers.go b/internal/http/memory_handlers.go index d35f630f..a5c9e6ab 100644 --- a/internal/http/memory_handlers.go +++ b/internal/http/memory_handlers.go @@ -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) diff --git a/internal/store/pg/sessions_list.go b/internal/store/pg/sessions_list.go index 2d574d30..6e795f68 100644 --- a/internal/store/pg/sessions_list.go +++ b/internal/store/pg/sessions_list.go @@ -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) { diff --git a/internal/store/sqlitestore/sessions_ops.go b/internal/store/sqlitestore/sessions_ops.go index ee98e581..ff547025 100644 --- a/internal/store/sqlitestore/sessions_ops.go +++ b/internal/store/sqlitestore/sessions_ops.go @@ -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) {