mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-11 03:13:24 +00:00
fix(memory): surface passive channel episodic recall (#1410)
* fix(memory): surface passive channel episodic recall * fix(memory): expose episodic key topics --------- Co-authored-by: Collective Developer <man@collective.dev>
This commit is contained in:
1 parent
12d5619cdc
commit
97e316f460
14 files changed
+315
-54
No files matched your search
@@ -22,19 +22,24 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string
|
||||
return nil, fmt.Errorf("item is not approvable")
|
||||
}
|
||||
sourceID := item.SourceID
|
||||
exists, err := s.Episodic.ExistsBySourceID(ctx, item.AgentID.String(), item.UserID, sourceID)
|
||||
// Passive channel extraction is reusable channel context, not a personal
|
||||
// memory of the channel instance creator or current sender.
|
||||
memoryUserID := ""
|
||||
exists, err := s.Episodic.ExistsBySourceID(ctx, item.AgentID.String(), memoryUserID, sourceID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !exists {
|
||||
retention := s.retentionDuration(ctx, item)
|
||||
keyTopics := memoryKeyTopics(item)
|
||||
ep := &store.EpisodicSummary{
|
||||
TenantID: item.TenantID,
|
||||
AgentID: item.AgentID,
|
||||
UserID: item.UserID,
|
||||
UserID: memoryUserID,
|
||||
SessionKey: "channel:" + item.ChannelInstanceID.String(),
|
||||
Summary: item.Summary,
|
||||
KeyTopics: decodeStrings(item.Topics),
|
||||
KeyTopics: keyTopics,
|
||||
L0Abstract: item.Summary,
|
||||
SourceID: sourceID,
|
||||
SourceType: "channel",
|
||||
ExpiresAt: timePtr(time.Now().UTC().Add(retention)),
|
||||
@@ -54,13 +59,13 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string
|
||||
EpisodicID: ep.ID.String(),
|
||||
SessionKey: ep.SessionKey,
|
||||
Summary: item.Summary,
|
||||
KeyTopics: decodeStrings(item.Topics),
|
||||
KeyTopics: keyTopics,
|
||||
KeyEntities: decodeStrings(item.Entities),
|
||||
},
|
||||
})
|
||||
}
|
||||
} else {
|
||||
ep, err := s.Episodic.GetBySourceID(ctx, item.AgentID.String(), item.UserID, sourceID)
|
||||
ep, err := s.Episodic.GetBySourceID(ctx, item.AgentID.String(), memoryUserID, sourceID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -170,13 +170,15 @@ func (f *fakeChannelStore) Get(context.Context, uuid.UUID) (*store.ChannelInstan
|
||||
|
||||
type fakeEpisodicStore struct {
|
||||
store.EpisodicStore
|
||||
exists bool
|
||||
bySource *store.EpisodicSummary
|
||||
created []*store.EpisodicSummary
|
||||
getCalls int
|
||||
createErr error
|
||||
existsErr error
|
||||
getByErr error
|
||||
exists bool
|
||||
bySource *store.EpisodicSummary
|
||||
created []*store.EpisodicSummary
|
||||
getCalls int
|
||||
existsUserID string
|
||||
getUserID string
|
||||
createErr error
|
||||
existsErr error
|
||||
getByErr error
|
||||
}
|
||||
|
||||
type fakeDomainEventBus struct {
|
||||
@@ -207,12 +209,14 @@ func (f *fakeEpisodicStore) Create(_ context.Context, ep *store.EpisodicSummary)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeEpisodicStore) ExistsBySourceID(context.Context, string, string, string) (bool, error) {
|
||||
func (f *fakeEpisodicStore) ExistsBySourceID(_ context.Context, _, userID, _ string) (bool, error) {
|
||||
f.existsUserID = userID
|
||||
return f.exists, f.existsErr
|
||||
}
|
||||
|
||||
func (f *fakeEpisodicStore) GetBySourceID(context.Context, string, string, string) (*store.EpisodicSummary, error) {
|
||||
func (f *fakeEpisodicStore) GetBySourceID(_ context.Context, _, userID, _ string) (*store.EpisodicSummary, error) {
|
||||
f.getCalls++
|
||||
f.getUserID = userID
|
||||
if f.getByErr != nil {
|
||||
return nil, f.getByErr
|
||||
}
|
||||
@@ -621,6 +625,12 @@ func TestApproveUsesConfiguredRetentionHours(t *testing.T) {
|
||||
if len(episodic.created) != 1 {
|
||||
t.Fatalf("created episodic count = %d, want 1", len(episodic.created))
|
||||
}
|
||||
if episodic.created[0].UserID != "" {
|
||||
t.Fatalf("channel episodic user_id = %q, want shared scope", episodic.created[0].UserID)
|
||||
}
|
||||
if episodic.created[0].L0Abstract != extractions.items[itemID].Summary {
|
||||
t.Fatalf("channel episodic L0Abstract = %q, want summary", episodic.created[0].L0Abstract)
|
||||
}
|
||||
expires := episodic.created[0].ExpiresAt
|
||||
if expires == nil {
|
||||
t.Fatal("episodic ExpiresAt is nil")
|
||||
@@ -667,7 +677,10 @@ func TestApprovePublishesTopicsAndEntitiesForSemanticHints(t *testing.T) {
|
||||
if len(episodic.created) != 1 {
|
||||
t.Fatalf("created episodic count = %d, want 1", len(episodic.created))
|
||||
}
|
||||
if got := episodic.created[0].KeyTopics; strings.Join(got, ",") != "collaboration,planning" {
|
||||
if episodic.existsUserID != "" {
|
||||
t.Fatalf("ExistsBySourceID user_id = %q, want shared scope", episodic.existsUserID)
|
||||
}
|
||||
if got := strings.Join(episodic.created[0].KeyTopics, ","); got != "collaboration,planning,Project Orion,ExampleCo" {
|
||||
t.Fatalf("episodic key_topics = %#v", got)
|
||||
}
|
||||
if len(eventBus.published) != 1 {
|
||||
@@ -677,7 +690,7 @@ func TestApprovePublishesTopicsAndEntitiesForSemanticHints(t *testing.T) {
|
||||
if !ok {
|
||||
t.Fatalf("published payload type = %T", eventBus.published[0].Payload)
|
||||
}
|
||||
if got := strings.Join(payload.KeyTopics, ","); got != "collaboration,planning" {
|
||||
if got := strings.Join(payload.KeyTopics, ","); got != "collaboration,planning,Project Orion,ExampleCo" {
|
||||
t.Fatalf("payload KeyTopics = %q", got)
|
||||
}
|
||||
if got := strings.Join(payload.KeyEntities, ","); got != "Project Orion,ExampleCo" {
|
||||
@@ -716,6 +729,9 @@ func TestApproveExistingSourceWritesExistingEpisodicID(t *testing.T) {
|
||||
if episodic.getCalls != 1 {
|
||||
t.Fatalf("GetBySourceID calls = %d, want 1", episodic.getCalls)
|
||||
}
|
||||
if episodic.getUserID != "" {
|
||||
t.Fatalf("GetBySourceID user_id = %q, want shared scope", episodic.getUserID)
|
||||
}
|
||||
if item.EpisodicID != existingID.String() {
|
||||
t.Fatalf("item episodic_id = %q, want %q", item.EpisodicID, existingID)
|
||||
}
|
||||
|
||||
@@ -62,6 +62,33 @@ func decodeStrings(raw json.RawMessage) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
func memoryKeyTopics(item *store.ChannelMemoryExtractionItem) []string {
|
||||
if item == nil {
|
||||
return nil
|
||||
}
|
||||
return mergeTopicLabels(decodeStrings(item.Topics), decodeStrings(item.Entities))
|
||||
}
|
||||
|
||||
func mergeTopicLabels(groups ...[]string) []string {
|
||||
seen := make(map[string]struct{})
|
||||
var out []string
|
||||
for _, group := range groups {
|
||||
for _, value := range group {
|
||||
label := strings.TrimSpace(value)
|
||||
if label == "" {
|
||||
continue
|
||||
}
|
||||
key := strings.ToLower(label)
|
||||
if _, ok := seen[key]; ok {
|
||||
continue
|
||||
}
|
||||
seen[key] = struct{}{}
|
||||
out = append(out, label)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
//go:fix inline
|
||||
func timePtr(t time.Time) *time.Time { return &t }
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ type EpisodicSummary struct {
|
||||
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"`
|
||||
|
||||
@@ -19,6 +19,7 @@ type episodicScored struct {
|
||||
id string
|
||||
sessionKey string
|
||||
l0 string
|
||||
keyTopics []string
|
||||
score float64
|
||||
createdAt time.Time
|
||||
}
|
||||
@@ -27,7 +28,7 @@ type episodicScored struct {
|
||||
// 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, l0_abstract,
|
||||
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
|
||||
@@ -35,10 +36,14 @@ func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID
|
||||
args := []any{query, agentID}
|
||||
p := 3
|
||||
|
||||
if userID != "" {
|
||||
q += fmt.Sprintf(" AND user_id = $%d", p)
|
||||
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))
|
||||
@@ -61,17 +66,22 @@ func (s *PGEpisodicStore) ftsSearch(ctx context.Context, query, agentID, userID
|
||||
// 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 {
|
||||
vecStr := vectorToString(embedding)
|
||||
q := `SELECT id, session_key, l0_abstract, 1 - (embedding <=> $1) AS score, created_at
|
||||
q := `SELECT id, session_key, COALESCE(NULLIF(l0_abstract, ''), left(summary, 500)) AS l0_abstract,
|
||||
key_topics, 1 - (embedding <=> $1) AS score, created_at
|
||||
FROM episodic_summaries
|
||||
WHERE agent_id = $2
|
||||
AND embedding IS NOT NULL`
|
||||
args := []any{vecStr, agentID}
|
||||
p := 3
|
||||
|
||||
if userID != "" {
|
||||
q += fmt.Sprintf(" AND user_id = $%d", p)
|
||||
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))
|
||||
@@ -94,13 +104,13 @@ func (s *PGEpisodicStore) vectorSearch(ctx context.Context, embedding []float32,
|
||||
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, 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, 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, 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, score: r.score * vecWeight}
|
||||
}
|
||||
}
|
||||
var merged []episodicScored
|
||||
|
||||
@@ -31,6 +31,9 @@ func (s *PGEpisodicStore) Close() error { retu
|
||||
func (s *PGEpisodicStore) Create(ctx context.Context, ep *store.EpisodicSummary) error {
|
||||
id := uuid.Must(uuid.NewV7())
|
||||
ep.ID = id
|
||||
if ep.L0Abstract == "" {
|
||||
ep.L0Abstract = fallbackEpisodicL0(ep.Summary)
|
||||
}
|
||||
|
||||
topics := pq.Array(ep.KeyTopics)
|
||||
now := time.Now().UTC()
|
||||
@@ -63,6 +66,15 @@ func (s *PGEpisodicStore) Create(ctx context.Context, ep *store.EpisodicSummary)
|
||||
return nil
|
||||
}
|
||||
|
||||
func fallbackEpisodicL0(summary string) string {
|
||||
const maxRunes = 500
|
||||
runes := []rune(summary)
|
||||
if len(runes) <= maxRunes {
|
||||
return summary
|
||||
}
|
||||
return string(runes[:maxRunes])
|
||||
}
|
||||
|
||||
// Get retrieves an episodic summary by ID.
|
||||
func (s *PGEpisodicStore) Get(ctx context.Context, id string) (*store.EpisodicSummary, error) {
|
||||
row := s.db.QueryRowContext(ctx, `
|
||||
@@ -163,7 +175,7 @@ func (s *PGEpisodicStore) Search(ctx context.Context, query, agentID, userID str
|
||||
continue
|
||||
}
|
||||
results = append(results, store.EpisodicSearchResult{
|
||||
EpisodicID: m.id, L0Abstract: m.l0, Score: m.score,
|
||||
EpisodicID: m.id, L0Abstract: m.l0, KeyTopics: m.keyTopics, Score: m.score,
|
||||
CreatedAt: m.createdAt, SessionKey: m.sessionKey,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -146,11 +146,12 @@ func (r *episodicSummaryRow) toEpisodicSummary() store.EpisodicSummary {
|
||||
|
||||
// episodicScoredRow is an sqlx scan struct for ftsSearch/vectorSearch in episodic_search.go.
|
||||
type episodicScoredRow struct {
|
||||
ID string `db:"id"`
|
||||
SessionKey string `db:"session_key"`
|
||||
L0Abstract string `db:"l0_abstract"`
|
||||
Score float64 `db:"score"`
|
||||
CreatedAt time.Time `db:"created_at"`
|
||||
ID string `db:"id"`
|
||||
SessionKey string `db:"session_key"`
|
||||
L0Abstract string `db:"l0_abstract"`
|
||||
KeyTopics pq.StringArray `db:"key_topics"`
|
||||
Score float64 `db:"score"`
|
||||
CreatedAt time.Time `db:"created_at"`
|
||||
}
|
||||
|
||||
func (r *episodicScoredRow) toEpisodicScored() episodicScored {
|
||||
@@ -158,6 +159,7 @@ func (r *episodicScoredRow) toEpisodicScored() episodicScored {
|
||||
id: r.ID,
|
||||
sessionKey: r.SessionKey,
|
||||
l0: r.L0Abstract,
|
||||
keyTopics: []string(r.KeyTopics),
|
||||
score: r.Score,
|
||||
createdAt: r.CreatedAt,
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package pg
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -195,12 +196,15 @@ func TestEpisodicSummaryRow_ToEpisodicSummary(t *testing.T) {
|
||||
func TestEpisodicScoredRow_ToEpisodicScored(t *testing.T) {
|
||||
created := time.Now()
|
||||
r := episodicScoredRow{
|
||||
ID: "ep-1", SessionKey: "s-1", L0Abstract: "abs", Score: 0.5, CreatedAt: created,
|
||||
ID: "ep-1", SessionKey: "s-1", L0Abstract: "abs", KeyTopics: pq.StringArray{"topic-a", "Entity B"}, Score: 0.5, CreatedAt: created,
|
||||
}
|
||||
got := r.toEpisodicScored()
|
||||
if got.id != "ep-1" || got.sessionKey != "s-1" || got.l0 != "abs" || got.score != 0.5 {
|
||||
t.Errorf("%+v", got)
|
||||
}
|
||||
if strings.Join(got.keyTopics, ",") != "topic-a,Entity B" {
|
||||
t.Errorf("keyTopics = %v", got.keyTopics)
|
||||
}
|
||||
if !got.createdAt.Equal(created) {
|
||||
t.Errorf("createdAt mismatch")
|
||||
}
|
||||
@@ -722,12 +726,12 @@ func TestCustomSkillExportRow_EmptyOptionals(t *testing.T) {
|
||||
|
||||
func TestMergeEpisodicScores_WeightingAndMerge(t *testing.T) {
|
||||
fts := []episodicScored{
|
||||
{id: "a", sessionKey: "s-a", l0: "A", score: 1.0},
|
||||
{id: "b", sessionKey: "s-b", l0: "B", score: 0.5},
|
||||
{id: "a", sessionKey: "s-a", l0: "A", keyTopics: []string{"topic-a"}, score: 1.0},
|
||||
{id: "b", sessionKey: "s-b", l0: "B", keyTopics: []string{"topic-b"}, score: 0.5},
|
||||
}
|
||||
vec := []episodicScored{
|
||||
{id: "a", sessionKey: "s-a", l0: "A", score: 0.8},
|
||||
{id: "c", sessionKey: "s-c", l0: "C", score: 0.9},
|
||||
{id: "a", sessionKey: "s-a", l0: "A", keyTopics: []string{"topic-a"}, score: 0.8},
|
||||
{id: "c", sessionKey: "s-c", l0: "C", keyTopics: []string{"topic-c"}, score: 0.9},
|
||||
}
|
||||
merged := mergeEpisodicScores(fts, vec, 0.5, 0.5)
|
||||
if len(merged) != 3 {
|
||||
@@ -737,6 +741,13 @@ func TestMergeEpisodicScores_WeightingAndMerge(t *testing.T) {
|
||||
for _, r := range merged {
|
||||
byID[r.id] = r.score
|
||||
}
|
||||
byTopics := make(map[string]string)
|
||||
for _, r := range merged {
|
||||
byTopics[r.id] = strings.Join(r.keyTopics, ",")
|
||||
}
|
||||
if byTopics["a"] != "topic-a" || byTopics["c"] != "topic-c" {
|
||||
t.Errorf("topics not preserved: %v", byTopics)
|
||||
}
|
||||
// a: in both → 1.0*0.5 + 0.8*0.5 = 0.5 + 0.4 = 0.9
|
||||
if got := byID["a"]; got < 0.89 || got > 0.91 {
|
||||
t.Errorf("a score = %v, want ~0.9", got)
|
||||
|
||||
@@ -34,6 +34,9 @@ func (s *SQLiteEpisodicStore) Close() error { return nil }
|
||||
func (s *SQLiteEpisodicStore) Create(ctx context.Context, ep *store.EpisodicSummary) error {
|
||||
id := uuid.Must(uuid.NewV7())
|
||||
ep.ID = id
|
||||
if ep.L0Abstract == "" {
|
||||
ep.L0Abstract = fallbackEpisodicL0(ep.Summary)
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
|
||||
topics := jsonStringArray(ep.KeyTopics)
|
||||
@@ -63,6 +66,15 @@ func (s *SQLiteEpisodicStore) Create(ctx context.Context, ep *store.EpisodicSumm
|
||||
return nil
|
||||
}
|
||||
|
||||
func fallbackEpisodicL0(summary string) string {
|
||||
const maxRunes = 500
|
||||
runes := []rune(summary)
|
||||
if len(runes) <= maxRunes {
|
||||
return summary
|
||||
}
|
||||
return string(runes[:maxRunes])
|
||||
}
|
||||
|
||||
// Get retrieves an episodic summary by ID.
|
||||
func (s *SQLiteEpisodicStore) Get(ctx context.Context, id string) (*store.EpisodicSummary, error) {
|
||||
tenantID := tenantIDForInsert(ctx)
|
||||
|
||||
@@ -27,14 +27,24 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
|
||||
tenantID := tenantIDForInsert(ctx)
|
||||
pattern := "%" + escapeLike(query) + "%"
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, `
|
||||
SELECT id, l0_abstract, key_topics, created_at, session_key
|
||||
q := `
|
||||
SELECT id, COALESCE(NULLIF(l0_abstract, ''), substr(summary, 1, 500)) AS l0_abstract, key_topics, created_at, session_key
|
||||
FROM episodic_summaries
|
||||
WHERE agent_id = ? AND user_id = ? AND tenant_id = ?
|
||||
AND (summary LIKE ? ESCAPE '\' OR key_topics LIKE ? ESCAPE '\')
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ?`,
|
||||
agentID, userID, tenantID.String(), pattern, pattern, maxResults*3)
|
||||
WHERE agent_id = ? AND tenant_id = ?
|
||||
AND (summary LIKE ? ESCAPE '\' OR key_topics LIKE ? ESCAPE '\')`
|
||||
args := []any{agentID, tenantID.String(), pattern, pattern}
|
||||
if store.IsSharedMemory(ctx) {
|
||||
// Shared memory searches all user scopes for the agent.
|
||||
} else if userID != "" {
|
||||
q += " AND (user_id = ? OR user_id = '')"
|
||||
args = append(args, userID)
|
||||
} else {
|
||||
q += " AND user_id = ''"
|
||||
}
|
||||
q += " ORDER BY created_at DESC LIMIT ?"
|
||||
args = append(args, maxResults*3)
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -94,6 +104,7 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
|
||||
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,
|
||||
SessionKey: sr.raw.sessionKey,
|
||||
@@ -105,5 +116,11 @@ func (s *SQLiteEpisodicStore) Search(ctx context.Context, query string, agentID,
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func searchKeyTopics(raw string) []string {
|
||||
var topics []string
|
||||
scanJSONStringArray([]byte(raw), &topics)
|
||||
return topics
|
||||
}
|
||||
|
||||
// Ensure SQLiteEpisodicStore implements store.EpisodicStore.
|
||||
var _ store.EpisodicStore = (*SQLiteEpisodicStore)(nil)
|
||||
@@ -15,10 +15,10 @@ import (
|
||||
|
||||
// MemorySearchTool implements the memory_search tool for hybrid semantic + FTS search.
|
||||
type MemorySearchTool struct {
|
||||
memStore store.MemoryStore // Postgres-backed
|
||||
episodicStore store.EpisodicStore // v3 episodic memory (nil = v2 fallback)
|
||||
metricsStore store.EvolutionMetricsStore // evolution metrics (nil = disabled)
|
||||
hasKG bool // knowledge_graph_search tool is available
|
||||
memStore store.MemoryStore // Postgres-backed
|
||||
episodicStore store.EpisodicStore // v3 episodic memory (nil = v2 fallback)
|
||||
metricsStore store.EvolutionMetricsStore // evolution metrics (nil = disabled)
|
||||
hasKG bool // knowledge_graph_search tool is available
|
||||
}
|
||||
|
||||
func NewMemorySearchTool() *MemorySearchTool {
|
||||
@@ -150,8 +150,9 @@ 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"`
|
||||
EpisodicID string `json:"episodic_id,omitempty"`
|
||||
L0 string `json:"l0_abstract,omitempty"`
|
||||
KeyTopics []string `json:"key_topics,omitempty"`
|
||||
EpisodicID string `json:"episodic_id,omitempty"`
|
||||
}
|
||||
var combined []taggedResult
|
||||
for _, r := range results {
|
||||
@@ -159,7 +160,7 @@ func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Re
|
||||
}
|
||||
for _, r := range episodicResults {
|
||||
combined = append(combined, taggedResult{
|
||||
Tier: "episodic", EpisodicID: r.EpisodicID, L0: r.L0Abstract,
|
||||
Tier: "episodic", EpisodicID: r.EpisodicID, L0: r.L0Abstract, KeyTopics: r.KeyTopics,
|
||||
MemorySearchResult: store.MemorySearchResult{
|
||||
Path: "episodic:" + r.SessionKey, Score: r.Score, Snippet: r.L0Abstract, Source: "episodic",
|
||||
},
|
||||
|
||||
@@ -3,6 +3,7 @@ package tools
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
@@ -23,8 +24,8 @@ func (t *MemoryExpandTool) SetEpisodicStore(es store.EpisodicStore) {
|
||||
t.episodicStore = es
|
||||
}
|
||||
|
||||
func (t *MemoryExpandTool) Name() string { return "memory_expand" }
|
||||
func (t *MemoryExpandTool) Description() string {
|
||||
func (t *MemoryExpandTool) Name() string { return "memory_expand" }
|
||||
func (t *MemoryExpandTool) Description() string {
|
||||
return "Load full content for a memory entry by ID. Returns the complete episodic summary for deep context."
|
||||
}
|
||||
|
||||
@@ -61,9 +62,12 @@ func (t *MemoryExpandTool) Execute(ctx context.Context, args map[string]any) *Re
|
||||
}
|
||||
|
||||
// Format full summary with metadata
|
||||
result := fmt.Sprintf("## Memory: %s\n\n**Session:** %s\n**Created:** %s\n**Turns:** %d\n\n%s",
|
||||
ep.L0Abstract, ep.SessionKey, ep.CreatedAt.Format("2006-01-02 15:04"),
|
||||
ep.TurnCount, ep.Summary)
|
||||
result := fmt.Sprintf("## Memory: %s\n\n**Session:** %s\n**Created:** %s\n**Turns:** %d",
|
||||
ep.L0Abstract, ep.SessionKey, ep.CreatedAt.Format("2006-01-02 15:04"), ep.TurnCount)
|
||||
if len(ep.KeyTopics) > 0 {
|
||||
result += "\n**Topics:** " + strings.Join(ep.KeyTopics, ", ")
|
||||
}
|
||||
result += "\n\n" + ep.Summary
|
||||
|
||||
return &Result{ForLLM: result}
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/store"
|
||||
)
|
||||
|
||||
type memorySearchFakeMemoryStore struct {
|
||||
store.MemoryStore
|
||||
results []store.MemorySearchResult
|
||||
}
|
||||
|
||||
func (f *memorySearchFakeMemoryStore) Search(context.Context, string, string, string, store.MemorySearchOptions) ([]store.MemorySearchResult, error) {
|
||||
return f.results, nil
|
||||
}
|
||||
|
||||
type memorySearchFakeEpisodicStore struct {
|
||||
store.EpisodicStore
|
||||
results []store.EpisodicSearchResult
|
||||
ep *store.EpisodicSummary
|
||||
}
|
||||
|
||||
func (f *memorySearchFakeEpisodicStore) Search(context.Context, string, string, string, store.EpisodicSearchOptions) ([]store.EpisodicSearchResult, error) {
|
||||
return f.results, nil
|
||||
}
|
||||
|
||||
func (f *memorySearchFakeEpisodicStore) Get(context.Context, string) (*store.EpisodicSummary, error) {
|
||||
return f.ep, nil
|
||||
}
|
||||
|
||||
func (f *memorySearchFakeEpisodicStore) RecordRecall(context.Context, string, float64) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestMemorySearchIncludesEpisodicKeyTopics(t *testing.T) {
|
||||
tool := NewMemorySearchTool()
|
||||
tool.SetMemoryStore(&memorySearchFakeMemoryStore{})
|
||||
tool.SetEpisodicStore(&memorySearchFakeEpisodicStore{results: []store.EpisodicSearchResult{
|
||||
{
|
||||
EpisodicID: "ep-1",
|
||||
L0Abstract: "BUV website Workshop 5 typography contrast.",
|
||||
KeyTopics: []string{"typography", "BUV", "Workshop 5"},
|
||||
Score: 0.8,
|
||||
CreatedAt: time.Now(),
|
||||
SessionKey: "channel:design",
|
||||
},
|
||||
}})
|
||||
|
||||
ctx := store.WithUserID(store.WithAgentID(context.Background(), uuid.New()), "user-1")
|
||||
res := tool.Execute(ctx, map[string]any{"query": "BUV typography"})
|
||||
if res.IsError {
|
||||
t.Fatalf("unexpected error: %s", res.ForLLM)
|
||||
}
|
||||
|
||||
var output struct {
|
||||
Results []struct {
|
||||
Tier string `json:"tier"`
|
||||
EpisodicID string `json:"episodic_id"`
|
||||
KeyTopics []string `json:"key_topics"`
|
||||
} `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)
|
||||
}
|
||||
got := output.Results[0]
|
||||
if got.Tier != "episodic" || got.EpisodicID != "ep-1" {
|
||||
t.Fatalf("episodic result metadata = %+v", got)
|
||||
}
|
||||
if strings.Join(got.KeyTopics, ",") != "typography,BUV,Workshop 5" {
|
||||
t.Fatalf("key_topics = %#v", got.KeyTopics)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryExpandIncludesEpisodicKeyTopics(t *testing.T) {
|
||||
tool := NewMemoryExpandTool()
|
||||
tool.SetEpisodicStore(&memorySearchFakeEpisodicStore{ep: &store.EpisodicSummary{
|
||||
L0Abstract: "BUV website Workshop 5.",
|
||||
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.",
|
||||
}})
|
||||
|
||||
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") {
|
||||
t.Fatalf("topics missing from expanded memory: %s", res.ForLLM)
|
||||
}
|
||||
}
|
||||
@@ -210,6 +210,48 @@ func TestStoreEpisodic_FTSSearch(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreEpisodic_SearchSharedBlankL0FallsBackToSummary(t *testing.T) {
|
||||
db := testDB(t)
|
||||
tenantID, agentID := seedTenantAgent(t, db)
|
||||
ctx := tenantCtx(tenantID)
|
||||
s := newEpisodicStore(t)
|
||||
|
||||
summary := "BUV website Workshop 5 Brand Style Test focuses on typography weight contrast"
|
||||
ep := &store.EpisodicSummary{
|
||||
TenantID: tenantID,
|
||||
AgentID: agentID,
|
||||
UserID: "",
|
||||
SessionKey: "channel:discord",
|
||||
Summary: summary,
|
||||
KeyTopics: []string{"BUV-website", "workshop-5", "typography"},
|
||||
SourceType: "channel",
|
||||
SourceID: "channel-blank-l0-" + tenantID.String()[:8],
|
||||
}
|
||||
if err := s.Create(ctx, ep); err != nil {
|
||||
t.Fatalf("Create: %v", err)
|
||||
}
|
||||
|
||||
if _, err := db.ExecContext(ctx, `UPDATE episodic_summaries SET l0_abstract = '' WHERE id = $1`, ep.ID); err != nil {
|
||||
t.Fatalf("force blank l0: %v", err)
|
||||
}
|
||||
|
||||
results, err := s.Search(ctx, "BUV website typography", agentID.String(), "guild:discord:user:nam", store.EpisodicSearchOptions{
|
||||
MaxResults: 10,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Search: %v", err)
|
||||
}
|
||||
if len(results) == 0 {
|
||||
t.Fatal("Search returned 0 results, expected shared channel memory")
|
||||
}
|
||||
if results[0].L0Abstract == "" {
|
||||
t.Fatal("Search returned blank L0Abstract, want summary fallback")
|
||||
}
|
||||
if results[0].L0Abstract != summary {
|
||||
t.Fatalf("Search L0Abstract = %q, want summary fallback", results[0].L0Abstract)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStoreEpisodic_TenantIsolation(t *testing.T) {
|
||||
db := testDB(t)
|
||||
tenantA, agentA := seedTenantAgent(t, db)
|
||||
|
||||
Reference in new issue
Block a user