feat(memory): add episodic time filters to search (#1411)

Co-authored-by: Collective Developer <man@collective.dev>
This commit is contained in:
ナムandCollective Developer authored and GitHub committed 2026-07-10 00:55:13 +07:00
1 parent 97e316f460
commit e2af8b5f15
10 files changed
+545 -81

No files matched your search

+2 -2
View File
@@ -98,11 +98,11 @@ The agent loop also uses this metadata for parallel tool-call scheduling. Only r
| Tool | Description |
|---|---|
| `memory_search` | Search memory documents (BM25 + vector hybrid) — returns L1 abstracts |
| `memory_search` | Search memory documents and episodic memory — supports episodic `created_at` filters via `timeRange`, `createdAfter`, and `createdBefore` |
| `memory_get` | Retrieve a specific memory document by ID |
| `memory_expand` | Load full episodic memory content (L2 deep retrieval) |
Memory layers: L1 (`memory_search`) returns ranked abstracts; L2 (`memory_expand`) loads the full summary for a given episodic ID.
Memory layers: L1 (`memory_search`) returns ranked abstracts; L2 (`memory_expand`) loads the full summary for a given episodic ID. Use `query: "*"` with an episodic time filter to list recent episodic memories without running document search.
### Sessions (`group:sessions`)
+14 -10
View File
@@ -34,20 +34,24 @@ type EpisodicSummary struct {
// EpisodicSearchResult is a search hit with L0 summary.
type EpisodicSearchResult struct {
EpisodicID string `json:"episodic_id" db:"episodic_id"`
L0Abstract string `json:"l0_abstract" db:"l0_abstract"`
KeyTopics []string `json:"key_topics" db:"key_topics"`
Score float64 `json:"score" db:"score"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
SessionKey string `json:"session_key" db:"session_key"`
EpisodicID string `json:"episodic_id" db:"episodic_id"`
L0Abstract string `json:"l0_abstract" db:"l0_abstract"`
KeyTopics []string `json:"key_topics" db:"key_topics"`
Score float64 `json:"score" db:"score"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
ExpiresAt *time.Time `json:"expires_at,omitempty" db:"expires_at"`
SessionKey string `json:"session_key" db:"session_key"`
}
// EpisodicSearchOptions configures episodic search behavior.
type EpisodicSearchOptions struct {
MaxResults int
MinScore float64
VectorWeight float64
TextWeight float64
MaxResults int
MinScore float64
VectorWeight float64
TextWeight float64
CreatedAfter *time.Time
CreatedBefore *time.Time
IncludeExpired bool
}
// EpisodicStore manages Tier 2 episodic memory.
+65 -27
View File
@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"fmt"
"strings"
"time"
"github.com/google/uuid"
@@ -22,20 +23,15 @@ type episodicScored struct {
keyTopics []string
score float64
createdAt time.Time
expiresAt *time.Time
}
// ftsSearch performs full-text search on episodic summaries.
// Uses the stored search_vector column (GIN-indexed, 'english' config from migration 040).
// When userID is empty, returns results across all users (admin view).
func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID string, limit int) []episodicScored {
q := `SELECT id, session_key, COALESCE(NULLIF(l0_abstract, ''), left(summary, 500)) AS l0_abstract, key_topics,
ts_rank(search_vector, plainto_tsquery('english', $1)) AS score, created_at
FROM episodic_summaries
WHERE agent_id = $2
AND search_vector @@ plainto_tsquery('english', $1)`
args := []any{query, agentID}
p := 3
func isEpisodicListQuery(query string) bool {
q := strings.TrimSpace(query)
return q == "" || q == "*"
}
func appendEpisodicSearchFilters(ctx context.Context, q string, args []any, p int, userID string, opts store.EpisodicSearchOptions) (string, []any, int) {
if store.IsSharedMemory(ctx) {
// Shared memory searches all user scopes for the agent.
} else if userID != "" {
@@ -48,6 +44,37 @@ func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID
q += fmt.Sprintf(" AND tenant_id = $%d", p)
args = append(args, tenantFromCtx(ctx))
p++
if opts.CreatedAfter != nil {
q += fmt.Sprintf(" AND created_at >= $%d", p)
args = append(args, opts.CreatedAfter.UTC())
p++
}
if opts.CreatedBefore != nil {
q += fmt.Sprintf(" AND created_at < $%d", p)
args = append(args, opts.CreatedBefore.UTC())
p++
}
if !opts.IncludeExpired {
q += fmt.Sprintf(" AND (expires_at IS NULL OR expires_at > $%d)", p)
args = append(args, time.Now().UTC())
p++
}
return q, args, p
}
// ftsSearch performs full-text search on episodic summaries.
// Uses the stored search_vector column (GIN-indexed, 'english' config from migration 040).
// When userID is empty, returns results across all users (admin view).
func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID string, limit int, opts store.EpisodicSearchOptions) []episodicScored {
q := `SELECT id, session_key, COALESCE(NULLIF(l0_abstract, ''), left(summary, 500)) AS l0_abstract, key_topics,
ts_rank(search_vector, plainto_tsquery('english', $1)) AS score, created_at, expires_at
FROM episodic_summaries
WHERE agent_id = $2
AND search_vector @@ plainto_tsquery('english', $1)`
args := []any{query, agentID}
p := 3
q, args, p = appendEpisodicSearchFilters(ctx, q, args, p, userID, opts)
q += fmt.Sprintf(" ORDER BY score DESC LIMIT $%d", p)
args = append(args, limit)
@@ -64,28 +91,17 @@ func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID
// vectorSearch performs cosine similarity search on episodic embeddings.
// When userID is empty, returns results across all users (admin view).
func (s *PGEpisodicStore) vectorSearch(ctx context.Context, embedding []float32, agentID, userID string, limit int) []episodicScored {
func (s *PGEpisodicStore) vectorSearch(ctx context.Context, embedding []float32, agentID, userID string, limit int, opts store.EpisodicSearchOptions) []episodicScored {
vecStr := vectorToString(embedding)
q := `SELECT id, session_key, COALESCE(NULLIF(l0_abstract, ''), left(summary, 500)) AS l0_abstract,
key_topics, 1 - (embedding <=> $1) AS score, created_at
key_topics, 1 - (embedding <=> $1) AS score, created_at, expires_at
FROM episodic_summaries
WHERE agent_id = $2
AND embedding IS NOT NULL`
args := []any{vecStr, agentID}
p := 3
if store.IsSharedMemory(ctx) {
// Shared memory searches all user scopes for the agent.
} else if userID != "" {
q += fmt.Sprintf(" AND (user_id = $%d OR user_id = '')", p)
args = append(args, userID)
p++
} else {
q += " AND user_id = ''"
}
q += fmt.Sprintf(" AND tenant_id = $%d", p)
args = append(args, tenantFromCtx(ctx))
p++
q, args, p = appendEpisodicSearchFilters(ctx, q, args, p, userID, opts)
q += fmt.Sprintf(" ORDER BY embedding <=> $1 LIMIT $%d", p)
args = append(args, limit)
@@ -100,17 +116,39 @@ func (s *PGEpisodicStore) vectorSearch(ctx context.Context, embedding []float32,
return results
}
func (s *PGEpisodicStore) recentSearch(ctx context.Context, agentID, userID string, limit int, opts store.EpisodicSearchOptions) []episodicScored {
q := `SELECT id, session_key, COALESCE(NULLIF(l0_abstract, ''), left(summary, 500)) AS l0_abstract, key_topics,
1.0 AS score, created_at, expires_at
FROM episodic_summaries
WHERE agent_id = $1`
args := []any{agentID}
p := 2
q, args, p = appendEpisodicSearchFilters(ctx, q, args, p, userID, opts)
q += fmt.Sprintf(" ORDER BY created_at DESC LIMIT $%d", p)
args = append(args, limit)
var rows []episodicScoredRow
if err := pkgSqlxDB.SelectContext(ctx, &rows, q, args...); err != nil {
return nil
}
results := make([]episodicScored, len(rows))
for i := range rows {
results[i] = rows[i].toEpisodicScored()
}
return results
}
// mergeEpisodicScores merges FTS and vector results by combined weighted score.
func mergeEpisodicScores(fts, vec []episodicScored, textWeight, vecWeight float64) []episodicScored {
byID := make(map[string]*episodicScored)
for _, r := range fts {
byID[r.id] = &episodicScored{id: r.id, sessionKey: r.sessionKey, l0: r.l0, keyTopics: r.keyTopics, createdAt: r.createdAt, score: r.score * textWeight}
byID[r.id] = &episodicScored{id: r.id, sessionKey: r.sessionKey, l0: r.l0, keyTopics: r.keyTopics, createdAt: r.createdAt, expiresAt: r.expiresAt, score: r.score * textWeight}
}
for _, r := range vec {
if existing, ok := byID[r.id]; ok {
existing.score += r.score * vecWeight
} else {
byID[r.id] = &episodicScored{id: r.id, sessionKey: r.sessionKey, l0: r.l0, keyTopics: r.keyTopics, createdAt: r.createdAt, score: r.score * vecWeight}
byID[r.id] = &episodicScored{id: r.id, sessionKey: r.sessionKey, l0: r.l0, keyTopics: r.keyTopics, createdAt: r.createdAt, expiresAt: r.expiresAt, score: r.score * vecWeight}
}
}
var merged []episodicScored
+20 -3
View File
@@ -6,6 +6,7 @@ import (
"fmt"
"log/slog"
"sort"
"strings"
"time"
"github.com/google/uuid"
@@ -149,15 +150,31 @@ func (s *PGEpisodicStore) Search(ctx context.Context, query, agentID, userID str
tw = 0.4
}
query = strings.TrimSpace(query)
if isEpisodicListQuery(query) {
merged := s.recentSearch(ctx, agentID, userID, maxResults, opts)
results := make([]store.EpisodicSearchResult, 0, len(merged))
for _, m := range merged {
if opts.MinScore > 0 && m.score < opts.MinScore {
continue
}
results = append(results, store.EpisodicSearchResult{
EpisodicID: m.id, L0Abstract: m.l0, KeyTopics: m.keyTopics, Score: m.score,
CreatedAt: m.createdAt, ExpiresAt: m.expiresAt, SessionKey: m.sessionKey,
})
}
return results, nil
}
// FTS search
ftsResults := s.ftsSearch(ctx, query, agentID, userID, maxResults*2)
ftsResults := s.ftsSearch(ctx, query, agentID, userID, maxResults*2, opts)
// Vector search (if embedding provider available)
var vecResults []episodicScored
if s.embProvider != nil {
vecs, err := s.embProvider.Embed(ctx, []string{query})
if err == nil && len(vecs) > 0 {
vecResults = s.vectorSearch(ctx, vecs[0], agentID, userID, maxResults*2)
vecResults = s.vectorSearch(ctx, vecs[0], agentID, userID, maxResults*2, opts)
}
}
@@ -176,7 +193,7 @@ func (s *PGEpisodicStore) Search(ctx context.Context, query, agentID, userID str
}
results = append(results, store.EpisodicSearchResult{
EpisodicID: m.id, L0Abstract: m.l0, KeyTopics: m.keyTopics, Score: m.score,
CreatedAt: m.createdAt, SessionKey: m.sessionKey,
CreatedAt: m.createdAt, ExpiresAt: m.expiresAt, SessionKey: m.sessionKey,
})
}
return results, nil
+2
View File
@@ -152,6 +152,7 @@ type episodicScoredRow struct {
KeyTopics pq.StringArray `db:"key_topics"`
Score float64 `db:"score"`
CreatedAt time.Time `db:"created_at"`
ExpiresAt *time.Time `db:"expires_at"`
}
func (r *episodicScoredRow) toEpisodicScored() episodicScored {
@@ -162,5 +163,6 @@ func (r *episodicScoredRow) toEpisodicScored() episodicScored {
keyTopics: []string(r.KeyTopics),
score: r.Score,
createdAt: r.CreatedAt,
expiresAt: r.ExpiresAt,
}
}
+38 -11
View File
@@ -6,6 +6,7 @@ import (
"context"
"sort"
"strings"
"time"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
@@ -25,14 +26,19 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
}
tenantID := tenantIDForInsert(ctx)
pattern := "%" + escapeLike(query) + "%"
query = strings.TrimSpace(query)
listOnly := query == "" || query == "*"
q := `
SELECT id, COALESCE(NULLIF(l0_abstract, ''), substr(summary, 1, 500)) AS l0_abstract, key_topics, created_at, session_key
SELECT id, COALESCE(NULLIF(l0_abstract, ''), substr(summary, 1, 500)) AS l0_abstract, key_topics, created_at, expires_at, session_key
FROM episodic_summaries
WHERE agent_id = ? AND tenant_id = ?
AND (summary LIKE ? ESCAPE '\' OR key_topics LIKE ? ESCAPE '\')`
args := []any{agentID, tenantID.String(), pattern, pattern}
WHERE agent_id = ? AND tenant_id = ?`
args := []any{agentID, tenantID.String()}
if !listOnly {
pattern := "%" + escapeLike(query) + "%"
q += " AND (summary LIKE ? ESCAPE '\\' OR key_topics LIKE ? ESCAPE '\\')"
args = append(args, pattern, pattern)
}
if store.IsSharedMemory(ctx) {
// Shared memory searches all user scopes for the agent.
} else if userID != "" {
@@ -41,6 +47,18 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
} else {
q += " AND user_id = ''"
}
if opts.CreatedAfter != nil {
q += " AND created_at >= ?"
args = append(args, opts.CreatedAfter.UTC().Format(time.RFC3339Nano))
}
if opts.CreatedBefore != nil {
q += " AND created_at < ?"
args = append(args, opts.CreatedBefore.UTC().Format(time.RFC3339Nano))
}
if !opts.IncludeExpired {
q += " AND (expires_at IS NULL OR expires_at > ?)"
args = append(args, time.Now().UTC().Format(time.RFC3339Nano))
}
q += " ORDER BY created_at DESC LIMIT ?"
args = append(args, maxResults*3)
@@ -55,13 +73,14 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
l0Abstract string
keyTopics string
createdAt sqliteTime
expiresAt nullSqliteTime
sessionKey string
}
var raws []rawRow
for rows.Next() {
var r rawRow
if err := rows.Scan(&r.id, &r.l0Abstract, &r.keyTopics, &r.createdAt, &r.sessionKey); err != nil {
if err := rows.Scan(&r.id, &r.l0Abstract, &r.keyTopics, &r.createdAt, &r.expiresAt, &r.sessionKey); err != nil {
continue
}
raws = append(raws, r)
@@ -79,11 +98,13 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
scoredRows := make([]scored, 0, len(raws))
for _, r := range raws {
sc := 1.0
if strings.Contains(strings.ToLower(r.l0Abstract), lowerQuery) {
sc += 0.2
}
if strings.Contains(strings.ToLower(r.keyTopics), lowerQuery) {
sc += 0.1
if !listOnly {
if strings.Contains(strings.ToLower(r.l0Abstract), lowerQuery) {
sc += 0.2
}
if strings.Contains(strings.ToLower(r.keyTopics), lowerQuery) {
sc += 0.1
}
}
scoredRows = append(scoredRows, scored{raw: r, score: sc})
}
@@ -101,12 +122,18 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
if opts.MinScore > 0 && sr.score < opts.MinScore {
continue
}
var expiresAt *time.Time
if sr.raw.expiresAt.Valid {
t := sr.raw.expiresAt.Time
expiresAt = &t
}
results = append(results, store.EpisodicSearchResult{
EpisodicID: sr.raw.id,
L0Abstract: sr.raw.l0Abstract,
KeyTopics: searchKeyTopics(sr.raw.keyTopics),
Score: sr.score,
CreatedAt: sr.raw.createdAt.Time,
ExpiresAt: expiresAt,
SessionKey: sr.raw.sessionKey,
})
if len(results) >= maxResults {
@@ -0,0 +1,90 @@
//go:build sqlite || sqliteonly
package sqlitestore
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
func TestSQLiteEpisodicSearch_TimeFiltersAndExpiry(t *testing.T) {
db := openTestDB(t)
if err := EnsureSchema(db); err != nil {
t.Fatalf("EnsureSchema: %v", err)
}
s := NewSQLiteEpisodicStore(db)
tenantID := uuid.New()
agentID := uuid.New()
userID := "time-user"
ctx := store.WithTenantID(context.Background(), tenantID)
base := time.Date(2026, 7, 9, 12, 0, 0, 0, time.UTC)
if _, err := db.ExecContext(ctx, `INSERT INTO tenants (id, name, slug, status) VALUES (?, ?, ?, ?)`, tenantID.String(), "Test Tenant", "test-tenant", "active"); err != nil {
t.Fatalf("seed tenant: %v", err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO agents (id, agent_key, display_name, owner_id, model, tenant_id) VALUES (?, ?, ?, ?, ?, ?)`, agentID.String(), "test-agent", "Test Agent", "owner", "test-model", tenantID.String()); err != nil {
t.Fatalf("seed agent: %v", err)
}
items := []struct {
sourceID string
summary string
createdAt time.Time
expiresAt *time.Time
}{
{sourceID: "old", summary: "Project Alpha old planning note", createdAt: base.Add(-48 * time.Hour)},
{sourceID: "recent", summary: "Project Alpha recent planning note", createdAt: base.Add(-2 * time.Hour)},
{sourceID: "expired", summary: "Project Alpha expired planning note", createdAt: base.Add(-time.Hour), expiresAt: sqliteEpisodicTimePtr(base.Add(-30 * time.Minute))},
}
for _, item := range items {
ep := &store.EpisodicSummary{
TenantID: tenantID,
AgentID: agentID,
UserID: userID,
SessionKey: "time-sess",
Summary: item.summary,
KeyTopics: []string{"project-alpha", "planning"},
SourceType: "session",
SourceID: "time-filter-" + item.sourceID,
ExpiresAt: item.expiresAt,
}
if err := s.Create(ctx, ep); err != nil {
t.Fatalf("Create %s: %v", item.sourceID, err)
}
if _, err := db.ExecContext(ctx, `UPDATE episodic_summaries SET created_at = ? WHERE id = ?`, item.createdAt.UTC().Format(time.RFC3339Nano), ep.ID.String()); err != nil {
t.Fatalf("set created_at %s: %v", item.sourceID, err)
}
}
createdAfter := base.Add(-24 * time.Hour)
results, err := s.Search(ctx, "*", agentID.String(), userID, store.EpisodicSearchOptions{
MaxResults: 10,
CreatedAfter: &createdAfter,
})
if err != nil {
t.Fatalf("Search recent: %v", err)
}
if len(results) != 1 {
t.Fatalf("recent non-expired results = %d, want 1", len(results))
}
if results[0].L0Abstract != "Project Alpha recent planning note" {
t.Fatalf("recent result = %q, want non-expired recent note", results[0].L0Abstract)
}
results, err = s.Search(ctx, "planning", agentID.String(), userID, store.EpisodicSearchOptions{
MaxResults: 10,
CreatedAfter: &createdAfter,
IncludeExpired: true,
})
if err != nil {
t.Fatalf("Search include expired: %v", err)
}
if len(results) != 2 {
t.Fatalf("include expired results = %d, want 2", len(results))
}
}
func sqliteEpisodicTimePtr(t time.Time) *time.Time { return &t }
+132 -16
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"log/slog"
"strconv"
"strings"
"time"
@@ -72,15 +73,35 @@ func (t *MemorySearchTool) Parameters() map[string]any {
"description": "Result depth: l0 (abstracts only), l1 (overview), l2 (full content). Default: l1. Only affects episodic memories.",
"enum": []string{"l0", "l1", "l2"},
},
"timeRange": map[string]any{
"type": "string",
"description": "Optional episodic-only relative created_at filter such as 24h, 7d, or 2w. Use query='*' with timeRange to list recent episodic memories.",
},
"createdAfter": map[string]any{
"type": "string",
"description": "Optional episodic-only created_at lower bound as an RFC3339 timestamp.",
},
"createdBefore": map[string]any{
"type": "string",
"description": "Optional episodic-only created_at upper bound as an RFC3339 timestamp.",
},
"includeExpired": map[string]any{
"type": "boolean",
"description": "Whether episodic search should include memories whose expires_at is in the past. Default false.",
},
},
"required": []string{"query"},
}
}
func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Result {
query, _ := args["query"].(string)
if query == "" {
return ErrorResult("query parameter is required")
query = strings.TrimSpace(query)
epTime, err := parseEpisodicTimeFilters(args, time.Now().UTC())
if err != nil {
return ErrorResult(err.Error())
}
if query == "" && !epTime.active {
return ErrorResult("query parameter is required unless an episodic time filter is provided")
}
var maxResults int
@@ -118,23 +139,30 @@ func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Re
}
}
agentStr := agentID.String()
results, err := t.memStore.Search(ctx, query, agentStr, userID, searchOpts)
if err != nil {
return ErrorResult(fmt.Sprintf("memory search failed: %v", err))
}
// Fallback: also search leader's memory for team members and merge results.
if leaderID := LeaderAgentIDFromCtx(ctx); leaderID != "" && leaderID != agentStr {
leaderResults, lerr := t.memStore.Search(ctx, query, leaderID, userID, searchOpts)
if lerr != nil && userID != "" {
leaderResults, _ = t.memStore.Search(ctx, query, leaderID, "", searchOpts)
var results []store.MemorySearchResult
searchDocuments := query != "" && query != "*"
if searchDocuments {
var err error
results, err = t.memStore.Search(ctx, query, agentStr, userID, searchOpts)
if err != nil {
return ErrorResult(fmt.Sprintf("memory search failed: %v", err))
}
// Fallback: also search leader's memory for team members and merge results.
if leaderID := LeaderAgentIDFromCtx(ctx); leaderID != "" && leaderID != agentStr {
leaderResults, lerr := t.memStore.Search(ctx, query, leaderID, userID, searchOpts)
if lerr != nil && userID != "" {
leaderResults, _ = t.memStore.Search(ctx, query, leaderID, "", searchOpts)
}
results = append(results, leaderResults...)
}
results = append(results, leaderResults...)
}
// V3 episodic memory search (merge with document results)
var episodicResults []store.EpisodicSearchResult
if t.episodicStore != nil {
epOpts := store.EpisodicSearchOptions{
MaxResults: maxResults, MinScore: minScore, VectorWeight: 0.3, TextWeight: 0.7,
CreatedAfter: epTime.createdAfter, CreatedBefore: epTime.createdBefore,
IncludeExpired: epTime.includeExpired,
}
epResults, epErr := t.episodicStore.Search(ctx, query, agentStr, userID, epOpts)
if epErr == nil {
@@ -150,17 +178,21 @@ func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Re
type taggedResult struct {
Tier string `json:"tier"`
store.MemorySearchResult
L0 string `json:"l0_abstract,omitempty"`
KeyTopics []string `json:"key_topics,omitempty"`
EpisodicID string `json:"episodic_id,omitempty"`
L0 string `json:"l0_abstract,omitempty"`
KeyTopics []string `json:"key_topics,omitempty"`
EpisodicID string `json:"episodic_id,omitempty"`
CreatedAt *time.Time `json:"created_at,omitempty"`
ExpiresAt *time.Time `json:"expires_at,omitempty"`
}
var combined []taggedResult
for _, r := range results {
combined = append(combined, taggedResult{Tier: "document", MemorySearchResult: r})
}
for _, r := range episodicResults {
createdAt := r.CreatedAt
combined = append(combined, taggedResult{
Tier: "episodic", EpisodicID: r.EpisodicID, L0: r.L0Abstract, KeyTopics: r.KeyTopics,
CreatedAt: &createdAt, ExpiresAt: r.ExpiresAt,
MemorySearchResult: store.MemorySearchResult{
Path: "episodic:" + r.SessionKey, Score: r.Score, Snippet: r.L0Abstract, Source: "episodic",
},
@@ -189,6 +221,90 @@ func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Re
return NewResult(string(data))
}
type episodicTimeFilters struct {
createdAfter *time.Time
createdBefore *time.Time
includeExpired bool
active bool
}
func parseEpisodicTimeFilters(args map[string]any, now time.Time) (episodicTimeFilters, error) {
var filters episodicTimeFilters
filters.includeExpired = boolArg(args, "includeExpired", "include_expired")
if s := stringArg(args, "createdAfter", "created_after"); s != "" {
t, err := time.Parse(time.RFC3339Nano, s)
if err != nil {
return filters, fmt.Errorf("createdAfter must be RFC3339: %w", err)
}
t = t.UTC()
filters.createdAfter = &t
filters.active = true
}
if s := stringArg(args, "createdBefore", "created_before"); s != "" {
t, err := time.Parse(time.RFC3339Nano, s)
if err != nil {
return filters, fmt.Errorf("createdBefore must be RFC3339: %w", err)
}
t = t.UTC()
filters.createdBefore = &t
filters.active = true
}
if s := stringArg(args, "timeRange", "time_range"); s != "" {
d, err := parseMemoryTimeRange(s)
if err != nil {
return filters, err
}
if filters.createdAfter == nil {
t := now.Add(-d).UTC()
filters.createdAfter = &t
}
filters.active = true
}
return filters, nil
}
func stringArg(args map[string]any, names ...string) string {
for _, name := range names {
if v, ok := args[name].(string); ok {
return strings.TrimSpace(v)
}
}
return ""
}
func boolArg(args map[string]any, names ...string) bool {
for _, name := range names {
if v, ok := args[name].(bool); ok {
return v
}
}
return false
}
func parseMemoryTimeRange(value string) (time.Duration, error) {
v := strings.TrimSpace(strings.ToLower(value))
if v == "" {
return 0, fmt.Errorf("timeRange cannot be empty")
}
if d, err := time.ParseDuration(v); err == nil {
return d, nil
}
unit := v[len(v)-1]
number := strings.TrimSpace(v[:len(v)-1])
n, err := strconv.ParseFloat(number, 64)
if err != nil || n <= 0 {
return 0, fmt.Errorf("timeRange must be a positive duration like 24h, 7d, or 2w")
}
switch unit {
case 'd':
return time.Duration(n * float64(24*time.Hour)), nil
case 'w':
return time.Duration(n * float64(7*24*time.Hour)), nil
default:
return 0, fmt.Errorf("timeRange must use a supported unit: h, d, or w")
}
}
// recordEpisodicRecall schedules a best-effort RecordRecall update per
// episodic hit so DreamingWorker can prioritise useful entries. Runs in a
// background goroutine bounded by a 5s timeout so slow DBs can't leak.
+72 -9
View File
@@ -15,9 +15,11 @@ import (
type memorySearchFakeMemoryStore struct {
store.MemoryStore
results []store.MemorySearchResult
calls int
}
func (f *memorySearchFakeMemoryStore) Search(context.Context, string, string, string, store.MemorySearchOptions) ([]store.MemorySearchResult, error) {
f.calls++
return f.results, nil
}
@@ -25,9 +27,13 @@ type memorySearchFakeEpisodicStore struct {
store.EpisodicStore
results []store.EpisodicSearchResult
ep *store.EpisodicSummary
query string
opts store.EpisodicSearchOptions
}
func (f *memorySearchFakeEpisodicStore) Search(context.Context, string, string, string, store.EpisodicSearchOptions) ([]store.EpisodicSearchResult, error) {
func (f *memorySearchFakeEpisodicStore) Search(_ context.Context, query string, _ string, _ string, opts store.EpisodicSearchOptions) ([]store.EpisodicSearchResult, error) {
f.query = query
f.opts = opts
return f.results, nil
}
@@ -45,8 +51,8 @@ func TestMemorySearchIncludesEpisodicKeyTopics(t *testing.T) {
tool.SetEpisodicStore(&memorySearchFakeEpisodicStore{results: []store.EpisodicSearchResult{
{
EpisodicID: "ep-1",
L0Abstract: "BUV website Workshop 5 typography contrast.",
KeyTopics: []string{"typography", "BUV", "Workshop 5"},
L0Abstract: "Project Alpha workshop notes mention visual hierarchy.",
KeyTopics: []string{"visual-hierarchy", "Project Alpha", "Workshop Notes"},
Score: 0.8,
CreatedAt: time.Now(),
SessionKey: "channel:design",
@@ -54,7 +60,7 @@ func TestMemorySearchIncludesEpisodicKeyTopics(t *testing.T) {
}})
ctx := store.WithUserID(store.WithAgentID(context.Background(), uuid.New()), "user-1")
res := tool.Execute(ctx, map[string]any{"query": "BUV typography"})
res := tool.Execute(ctx, map[string]any{"query": "Project Alpha visual hierarchy"})
if res.IsError {
t.Fatalf("unexpected error: %s", res.ForLLM)
}
@@ -76,26 +82,83 @@ func TestMemorySearchIncludesEpisodicKeyTopics(t *testing.T) {
if got.Tier != "episodic" || got.EpisodicID != "ep-1" {
t.Fatalf("episodic result metadata = %+v", got)
}
if strings.Join(got.KeyTopics, ",") != "typography,BUV,Workshop 5" {
if strings.Join(got.KeyTopics, ",") != "visual-hierarchy,Project Alpha,Workshop Notes" {
t.Fatalf("key_topics = %#v", got.KeyTopics)
}
}
func TestMemorySearchTimeRangeListsEpisodicOnly(t *testing.T) {
createdAt := time.Date(2026, 7, 9, 17, 21, 29, 0, time.UTC)
expiresAt := createdAt.Add(7 * 24 * time.Hour)
mem := &memorySearchFakeMemoryStore{}
ep := &memorySearchFakeEpisodicStore{results: []store.EpisodicSearchResult{
{
EpisodicID: "ep-1",
L0Abstract: "Project Alpha workshop notes mention visual hierarchy.",
KeyTopics: []string{"visual-hierarchy", "Project Alpha"},
Score: 1,
CreatedAt: createdAt,
ExpiresAt: &expiresAt,
SessionKey: "channel:design",
},
}}
tool := NewMemorySearchTool()
tool.SetMemoryStore(mem)
tool.SetEpisodicStore(ep)
ctx := store.WithUserID(store.WithAgentID(context.Background(), uuid.New()), "user-1")
res := tool.Execute(ctx, map[string]any{"query": "*", "timeRange": "24h"})
if res.IsError {
t.Fatalf("unexpected error: %s", res.ForLLM)
}
if mem.calls != 0 {
t.Fatalf("document memory search calls = %d, want 0 for query='*'", mem.calls)
}
if ep.query != "*" {
t.Fatalf("episodic query = %q, want *", ep.query)
}
if ep.opts.CreatedAfter == nil {
t.Fatal("CreatedAfter was not forwarded")
}
if ep.opts.IncludeExpired {
t.Fatal("IncludeExpired default = true, want false")
}
var output struct {
Results []struct {
CreatedAt string `json:"created_at"`
ExpiresAt *string `json:"expires_at"`
} `json:"results"`
}
if err := json.Unmarshal([]byte(res.ForLLM), &output); err != nil {
t.Fatalf("invalid JSON output: %v\n%s", err, res.ForLLM)
}
if len(output.Results) != 1 {
t.Fatalf("results = %d, want 1: %s", len(output.Results), res.ForLLM)
}
if output.Results[0].CreatedAt == "" {
t.Fatalf("created_at missing: %s", res.ForLLM)
}
if output.Results[0].ExpiresAt == nil || *output.Results[0].ExpiresAt == "" {
t.Fatalf("expires_at missing: %s", res.ForLLM)
}
}
func TestMemoryExpandIncludesEpisodicKeyTopics(t *testing.T) {
tool := NewMemoryExpandTool()
tool.SetEpisodicStore(&memorySearchFakeEpisodicStore{ep: &store.EpisodicSummary{
L0Abstract: "BUV website Workshop 5.",
L0Abstract: "Project Alpha workshop notes.",
SessionKey: "channel:design",
CreatedAt: time.Date(2026, 7, 9, 14, 55, 0, 0, time.UTC),
KeyTopics: []string{"typography", "BUV", "Workshop 5"},
Summary: "Typography weight contrast is a focus area.",
KeyTopics: []string{"visual-hierarchy", "Project Alpha", "Workshop Notes"},
Summary: "Visual hierarchy is a focus area.",
}})
res := tool.Execute(context.Background(), map[string]any{"id": "ep-1"})
if res.IsError {
t.Fatalf("unexpected error: %s", res.ForLLM)
}
if !strings.Contains(res.ForLLM, "**Topics:** typography, BUV, Workshop 5") {
if !strings.Contains(res.ForLLM, "**Topics:** visual-hierarchy, Project Alpha, Workshop Notes") {
t.Fatalf("topics missing from expanded memory: %s", res.ForLLM)
}
}
+110 -3
View File
@@ -19,6 +19,8 @@ func newEpisodicStore(t *testing.T) *pg.PGEpisodicStore {
return pg.NewPGEpisodicStore(db)
}
func episodicTimePtr(t time.Time) *time.Time { return &t }
func TestStoreEpisodic_CreateAndGet(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
@@ -216,14 +218,14 @@ func TestStoreEpisodic_SearchSharedBlankL0FallsBackToSummary(t *testing.T) {
ctx := tenantCtx(tenantID)
s := newEpisodicStore(t)
summary := "BUV website Workshop 5 Brand Style Test focuses on typography weight contrast"
summary := "Project Alpha workshop notes mention visual hierarchy as a focus area"
ep := &store.EpisodicSummary{
TenantID: tenantID,
AgentID: agentID,
UserID: "",
SessionKey: "channel:discord",
Summary: summary,
KeyTopics: []string{"BUV-website", "workshop-5", "typography"},
KeyTopics: []string{"project-alpha", "workshop-notes", "visual-hierarchy"},
SourceType: "channel",
SourceID: "channel-blank-l0-" + tenantID.String()[:8],
}
@@ -235,7 +237,7 @@ func TestStoreEpisodic_SearchSharedBlankL0FallsBackToSummary(t *testing.T) {
t.Fatalf("force blank l0: %v", err)
}
results, err := s.Search(ctx, "BUV website typography", agentID.String(), "guild:discord:user:nam", store.EpisodicSearchOptions{
results, err := s.Search(ctx, "Project Alpha visual hierarchy", agentID.String(), "guild:discord:user:sample", store.EpisodicSearchOptions{
MaxResults: 10,
})
if err != nil {
@@ -252,6 +254,111 @@ func TestStoreEpisodic_SearchSharedBlankL0FallsBackToSummary(t *testing.T) {
}
}
func TestStoreEpisodic_SearchCreatedAtAndExpiryFilters(t *testing.T) {
db := testDB(t)
tenantID, agentID := seedTenantAgent(t, db)
ctx := tenantCtx(tenantID)
s := newEpisodicStore(t)
userID := "time-user-" + tenantID.String()[:8]
base := time.Date(2026, 7, 9, 12, 0, 0, 0, time.UTC)
items := []struct {
sourceID string
summary string
createdAt time.Time
expiresAt *time.Time
}{
{
sourceID: "old",
summary: "Project Alpha old planning note",
createdAt: base.Add(-48 * time.Hour),
},
{
sourceID: "recent",
summary: "Project Alpha recent planning note",
createdAt: base.Add(-2 * time.Hour),
},
{
sourceID: "expired",
summary: "Project Alpha expired planning note",
createdAt: base.Add(-1 * time.Hour),
expiresAt: episodicTimePtr(base.Add(-30 * time.Minute)),
},
}
for _, item := range items {
ep := &store.EpisodicSummary{
TenantID: tenantID,
AgentID: agentID,
UserID: userID,
SessionKey: "time-sess",
Summary: item.summary,
KeyTopics: []string{"project-alpha", "planning"},
SourceType: "session",
SourceID: "time-filter-" + item.sourceID + "-" + tenantID.String()[:8],
}
if item.expiresAt != nil {
ep.ExpiresAt = item.expiresAt
}
if err := s.Create(ctx, ep); err != nil {
t.Fatalf("Create %s: %v", item.sourceID, err)
}
if _, err := db.ExecContext(ctx, `UPDATE episodic_summaries SET created_at = $1 WHERE id = $2`, item.createdAt, ep.ID); err != nil {
t.Fatalf("set created_at %s: %v", item.sourceID, err)
}
}
createdAfter := base.Add(-24 * time.Hour)
results, err := s.Search(ctx, "Project Alpha planning", agentID.String(), userID, store.EpisodicSearchOptions{
MaxResults: 10,
CreatedAfter: &createdAfter,
IncludeExpired: true,
})
if err != nil {
t.Fatalf("Search with CreatedAfter: %v", err)
}
if len(results) != 2 {
t.Fatalf("CreatedAfter results = %d, want 2 recent+expired", len(results))
}
for _, r := range results {
if r.CreatedAt.Before(createdAfter) {
t.Fatalf("result before CreatedAfter: %s < %s", r.CreatedAt, createdAfter)
}
}
results, err = s.Search(ctx, "*", agentID.String(), userID, store.EpisodicSearchOptions{
MaxResults: 10,
CreatedAfter: &createdAfter,
})
if err != nil {
t.Fatalf("Recent list search: %v", err)
}
if len(results) != 1 {
t.Fatalf("non-expired recent results = %d, want 1", len(results))
}
if results[0].L0Abstract != "Project Alpha recent planning note" {
t.Fatalf("recent result = %q, want non-expired recent note", results[0].L0Abstract)
}
if results[0].ExpiresAt != nil {
t.Fatalf("non-expired result ExpiresAt = %v, want nil", results[0].ExpiresAt)
}
createdBefore := base.Add(-24 * time.Hour)
results, err = s.Search(ctx, "*", agentID.String(), userID, store.EpisodicSearchOptions{
MaxResults: 10,
CreatedBefore: &createdBefore,
IncludeExpired: true,
})
if err != nil {
t.Fatalf("CreatedBefore list search: %v", err)
}
if len(results) != 1 {
t.Fatalf("CreatedBefore results = %d, want 1 old note", len(results))
}
if !results[0].CreatedAt.Before(createdBefore) {
t.Fatalf("result not before CreatedBefore: %s >= %s", results[0].CreatedAt, createdBefore)
}
}
func TestStoreEpisodic_TenantIsolation(t *testing.T) {
db := testDB(t)
tenantA, agentA := seedTenantAgent(t, db)