diff --git a/docs/03-tools-system.md b/docs/03-tools-system.md index 7d424a7d..9f8ab01e 100644 --- a/docs/03-tools-system.md +++ b/docs/03-tools-system.md @@ -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`) diff --git a/internal/store/episodic_store.go b/internal/store/episodic_store.go index c7daf18b..8e4b7cb6 100644 --- a/internal/store/episodic_store.go +++ b/internal/store/episodic_store.go @@ -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. diff --git a/internal/store/pg/episodic_search.go b/internal/store/pg/episodic_search.go index 06e421db..1ce5beaf 100644 --- a/internal/store/pg/episodic_search.go +++ b/internal/store/pg/episodic_search.go @@ -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 diff --git a/internal/store/pg/episodic_summaries.go b/internal/store/pg/episodic_summaries.go index 0f1c3b5d..e554fd97 100644 --- a/internal/store/pg/episodic_summaries.go +++ b/internal/store/pg/episodic_summaries.go @@ -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 diff --git a/internal/store/pg/memory_scan_rows.go b/internal/store/pg/memory_scan_rows.go index 5bb8886d..f22c8077 100644 --- a/internal/store/pg/memory_scan_rows.go +++ b/internal/store/pg/memory_scan_rows.go @@ -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, } } diff --git a/internal/store/sqlitestore/episodic_search.go b/internal/store/sqlitestore/episodic_search.go index 631d1594..d878ba60 100644 --- a/internal/store/sqlitestore/episodic_search.go +++ b/internal/store/sqlitestore/episodic_search.go @@ -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 { diff --git a/internal/store/sqlitestore/episodic_search_test.go b/internal/store/sqlitestore/episodic_search_test.go new file mode 100644 index 00000000..5d6f8133 --- /dev/null +++ b/internal/store/sqlitestore/episodic_search_test.go @@ -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 } diff --git a/internal/tools/memory.go b/internal/tools/memory.go index 1968755c..c6b8be1f 100644 --- a/internal/tools/memory.go +++ b/internal/tools/memory.go @@ -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. diff --git a/internal/tools/memory_search_topics_test.go b/internal/tools/memory_search_topics_test.go index 75d76635..2d7f08be 100644 --- a/internal/tools/memory_search_topics_test.go +++ b/internal/tools/memory_search_topics_test.go @@ -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) } } diff --git a/tests/integration/v3_episodic_store_test.go b/tests/integration/v3_episodic_store_test.go index 422946f8..215dc8a4 100644 --- a/tests/integration/v3_episodic_store_test.go +++ b/tests/integration/v3_episodic_store_test.go @@ -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)