From a9c7feb99cb7ae3b785a67ab40cbafd58c4a30c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=83=8A=E3=83=A0?= Date: Wed, 8 Jul 2026 02:25:08 +0800 Subject: [PATCH] fix(channelmemory): harden passive extraction reliability (#1386) Co-authored-by: Collective Developer --- docs/06-store-data-model.md | 4 +- docs/project-changelog.md | 17 + internal/channelmemory/config.go | 56 +++ internal/channelmemory/config_test.go | 16 + internal/channelmemory/extractor.go | 71 ++-- internal/channelmemory/review.go | 22 +- internal/channelmemory/service.go | 43 ++- internal/channelmemory/service_test.go | 328 +++++++++++++++++- internal/consolidation/workers_test.go | 9 + internal/http/channel_memory_extraction.go | 12 +- internal/store/episodic_store.go | 3 +- .../store/pg/channel_memory_extraction.go | 2 +- internal/store/pg/episodic_summaries.go | 13 +- .../sqlitestore/channel-memory-extraction.go | 2 +- .../channel_memory_extraction_test.go | 97 ++++++ internal/store/sqlitestore/episodic.go | 12 + internal/store/sqlitestore/schema.go | 33 +- internal/store/sqlitestore/schema.sql | 2 +- internal/upgrade/version.go | 2 +- .../000076_channel_memory_extraction.up.sql | 2 +- ...000090_channel_memory_item_dedupe.down.sql | 6 + .../000090_channel_memory_item_dedupe.up.sql | 36 ++ 22 files changed, 730 insertions(+), 58 deletions(-) create mode 100644 internal/store/sqlitestore/channel_memory_extraction_test.go create mode 100644 migrations/000090_channel_memory_item_dedupe.down.sql create mode 100644 migrations/000090_channel_memory_item_dedupe.up.sql diff --git a/docs/06-store-data-model.md b/docs/06-store-data-model.md index e2fcbae7..2c030ed7 100644 --- a/docs/06-store-data-model.md +++ b/docs/06-store-data-model.md @@ -720,7 +720,7 @@ L0 (Working Memory) L1 (Episodic Memory) L2 (Semantic Memory) | Interface | Purpose | Key Methods | |-----------|---------|-------------| -| `EpisodicStore` | Tier 1.5 memory CRUD + hybrid search | `Create`, `Search`, `ExistsBySourceID`, `ListUnpromoted`, `MarkPromoted` | +| `EpisodicStore` | Tier 1.5 memory CRUD + hybrid search | `Create`, `Search`, `ExistsBySourceID`, `GetBySourceID`, `ListUnpromoted`, `MarkPromoted` | | `EvolutionMetricsStore` | Stage 1: record metrics (retrieval, tool, feedback) | `RecordMetric`, `AggregateToolMetrics`, `AggregateRetrievalMetrics` | | `EvolutionSuggestionStore` | Stage 2: generate & track improvement suggestions | `CreateSuggestion`, `ListSuggestions`, `UpdateSuggestionStatus` | | `VaultStore` | Knowledge Vault: document registry + links | `UpsertDocument`, `Search`, `CreateLink`, `GetOutLinks`, `GetBacklinks` | @@ -744,7 +744,7 @@ L0 (Working Memory) L1 (Episodic Memory) L2 (Semantic Memory) `ChannelMemoryExtractionStore` is implemented for PostgreSQL and SQLite. It is tenant-scoped, stores no raw message bodies, and uses deterministic hashes to deduplicate the same channel/history/type/summary candidate across repeated -runs. +runs for the same channel instance. ### 12 Promoted Agent Columns diff --git a/docs/project-changelog.md b/docs/project-changelog.md index c7b27765..ed188151 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -4,6 +4,23 @@ Significant changes, features, and fixes in reverse chronological order. --- +## 2026-07-07 + +### Passive channel memory extraction reliability + +**Fixes** + +- Fixed partial settings updates so enabling passive memory no longer clears review mode or other omitted options. +- Extraction runs now checkpoint only messages actually sent within the extraction input budget, preventing skipped channel messages. +- Pending review counts now count the full queue, retention uses the configured `retention_hours`, duplicate approvals preserve the existing episodic ID, and failed item writes mark the run as failed. +- Passive memory candidates now dedupe across runs for the same channel instance and item hash in both PostgreSQL and SQLite. + +**Tests** + +- Added regression coverage for config patching, checkpoint bounds, pending counts, run failure finalization, retention, duplicate approval, and SQLite cross-run dedupe. + +--- + ## 2026-07-03 ### Usage cost display precision diff --git a/internal/channelmemory/config.go b/internal/channelmemory/config.go index 85baae67..7921b6fb 100644 --- a/internal/channelmemory/config.go +++ b/internal/channelmemory/config.go @@ -23,6 +23,20 @@ type Config struct { GroupOnly bool `json:"group_only"` } +type ConfigPatch struct { + Enabled *bool `json:"enabled"` + ReviewMode *bool `json:"review_mode"` + IntervalMinutes *int `json:"interval_minutes"` + MessageCap *int `json:"message_cap"` + RetentionHours *int `json:"retention_hours"` + AllowedTypes *[]string `json:"allowed_types"` + ExcludeUsers *[]string `json:"exclude_users"` + ExcludePatterns *[]string `json:"exclude_patterns"` + ExcludeHistoryKeys *[]string `json:"exclude_history_keys"` + MinMessages *int `json:"min_messages"` + GroupOnly *bool `json:"group_only"` +} + func DefaultConfig() Config { return Config{ Enabled: false, @@ -62,6 +76,48 @@ func ParseConfig(raw json.RawMessage) Config { return cfg } +func ApplyConfigPatch(base Config, raw []byte) (Config, error) { + var patch ConfigPatch + if err := json.Unmarshal(raw, &patch); err != nil { + return Config{}, err + } + cfg := base + if patch.Enabled != nil { + cfg.Enabled = *patch.Enabled + } + if patch.ReviewMode != nil { + cfg.ReviewMode = *patch.ReviewMode + } + if patch.IntervalMinutes != nil { + cfg.IntervalMinutes = *patch.IntervalMinutes + } + if patch.MessageCap != nil { + cfg.MessageCap = *patch.MessageCap + } + if patch.RetentionHours != nil { + cfg.RetentionHours = *patch.RetentionHours + } + if patch.AllowedTypes != nil { + cfg.AllowedTypes = *patch.AllowedTypes + } + if patch.ExcludeUsers != nil { + cfg.ExcludeUsers = *patch.ExcludeUsers + } + if patch.ExcludePatterns != nil { + cfg.ExcludePatterns = *patch.ExcludePatterns + } + if patch.ExcludeHistoryKeys != nil { + cfg.ExcludeHistoryKeys = *patch.ExcludeHistoryKeys + } + if patch.MinMessages != nil { + cfg.MinMessages = *patch.MinMessages + } + if patch.GroupOnly != nil { + cfg.GroupOnly = *patch.GroupOnly + } + return ParseConfig(MergeIntoInstanceConfig(nil, cfg)), nil +} + func MergeIntoInstanceConfig(raw json.RawMessage, cfg Config) json.RawMessage { var root map[string]any if len(raw) == 0 || json.Unmarshal(raw, &root) != nil || root == nil { diff --git a/internal/channelmemory/config_test.go b/internal/channelmemory/config_test.go index 1eee9215..922e4d03 100644 --- a/internal/channelmemory/config_test.go +++ b/internal/channelmemory/config_test.go @@ -83,3 +83,19 @@ func TestMergeIntoInstanceConfigPreservesSiblingFields(t *testing.T) { t.Fatalf("passive_memory missing: %s", raw) } } + +func TestApplyConfigPatchPreservesUnspecifiedReviewMode(t *testing.T) { + base := DefaultConfig() + base.ReviewMode = true + + cfg, err := ApplyConfigPatch(base, []byte(`{"enabled":true}`)) + if err != nil { + t.Fatal(err) + } + if !cfg.Enabled { + t.Fatal("enabled patch was not applied") + } + if !cfg.ReviewMode { + t.Fatal("partial patch must preserve review_mode") + } +} diff --git a/internal/channelmemory/extractor.go b/internal/channelmemory/extractor.go index c9685af2..a5416220 100644 --- a/internal/channelmemory/extractor.go +++ b/internal/channelmemory/extractor.go @@ -69,32 +69,12 @@ func callExtractionProvider( maxOutputTokens int, purpose string, ) (*providers.ChatResponse, error) { - var sb strings.Builder - for _, msg := range messages { - sb.WriteString(msg.CreatedAt.Format(time.RFC3339)) - sb.WriteString(" ") - if msg.Sender != "" { - sb.WriteString(msg.Sender) - } else { - sb.WriteString(msg.SenderID) - } - sb.WriteString(": ") - body := msg.Body - if len([]rune(body)) > 800 { - body = string([]rune(body)[:800]) + "..." - } - sb.WriteString(body) - sb.WriteByte('\n') - if sb.Len() > maxInputChars { - sb.WriteString("...(truncated)\n") - break - } - } + input := buildExtractionInput(messagesWithinExtractionBudget(messages, maxInputChars)) req := providers.ChatRequest{ Model: model, Messages: []providers.Message{ {Role: "system", Content: extractionPrompt(allowed)}, - {Role: "user", Content: sb.String()}, + {Role: "user", Content: input}, }, Options: map[string]any{"max_tokens": maxOutputTokens, "temperature": 0.1}, } @@ -108,6 +88,53 @@ func callExtractionProvider( return provider.Chat(ctx, req) } +func messagesWithinExtractionBudget(messages []store.PendingMessage, maxInputChars int) []store.PendingMessage { + if maxInputChars <= 0 { + return nil + } + var sb strings.Builder + out := make([]store.PendingMessage, 0, len(messages)) + for _, msg := range messages { + line := extractionMessageLine(msg) + if len(out) > 0 && sb.Len()+len(line) > maxInputChars { + break + } + if len(out) == 0 && len(line) > maxInputChars { + break + } + sb.WriteString(line) + out = append(out, msg) + } + return out +} + +func buildExtractionInput(messages []store.PendingMessage) string { + var sb strings.Builder + for _, msg := range messages { + sb.WriteString(extractionMessageLine(msg)) + } + return sb.String() +} + +func extractionMessageLine(msg store.PendingMessage) string { + var sb strings.Builder + sb.WriteString(msg.CreatedAt.Format(time.RFC3339)) + sb.WriteString(" ") + if msg.Sender != "" { + sb.WriteString(msg.Sender) + } else { + sb.WriteString(msg.SenderID) + } + sb.WriteString(": ") + body := msg.Body + if len([]rune(body)) > 800 { + body = string([]rune(body)[:800]) + "..." + } + sb.WriteString(body) + sb.WriteByte('\n') + return sb.String() +} + func extractionPrompt(allowed []string) string { return `Extract only durable, reusable work context from channel messages. Allowed item types: ` + strings.Join(allowed, ", ") + `. diff --git a/internal/channelmemory/review.go b/internal/channelmemory/review.go index 384b8225..722ec787 100644 --- a/internal/channelmemory/review.go +++ b/internal/channelmemory/review.go @@ -27,6 +27,7 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string return nil, err } if !exists { + retention := s.retentionDuration(ctx, item) ep := &store.EpisodicSummary{ TenantID: item.TenantID, AgentID: item.AgentID, @@ -36,7 +37,7 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string KeyTopics: decodeStrings(item.Topics), SourceID: sourceID, SourceType: "channel", - ExpiresAt: timePtr(time.Now().UTC().Add(90 * 24 * time.Hour)), + ExpiresAt: timePtr(time.Now().UTC().Add(retention)), } if err := s.Episodic.Create(ctx, ep); err != nil { return nil, err @@ -57,6 +58,14 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string }, }) } + } else { + ep, err := s.Episodic.GetBySourceID(ctx, item.AgentID.String(), item.UserID, sourceID) + if err != nil { + return nil, err + } + if ep != nil { + item.EpisodicID = ep.ID.String() + } } now := time.Now().UTC() if err := s.Extractions.UpdateItem(ctx, item.ID, map[string]any{ @@ -72,6 +81,17 @@ func (s *Service) Approve(ctx context.Context, itemID uuid.UUID, approver string return item, nil } +func (s *Service) retentionDuration(ctx context.Context, item *store.ChannelMemoryExtractionItem) time.Duration { + cfg := DefaultConfig() + if s.Channels != nil && item != nil { + inst, err := s.Channels.Get(ctx, item.ChannelInstanceID) + if err == nil && inst != nil { + cfg = ParseConfig(inst.Config) + } + } + return time.Duration(cfg.RetentionHours) * time.Hour +} + func (s *Service) Reject(ctx context.Context, itemID uuid.UUID, actor string) error { return s.updateItemTerminal(ctx, itemID, store.ChannelMemoryItemRejected, map[string]any{"rejected_by": actor, "rejected_at": time.Now().UTC()}) } diff --git a/internal/channelmemory/service.go b/internal/channelmemory/service.go index 3476bc9d..3b48c4fa 100644 --- a/internal/channelmemory/service.go +++ b/internal/channelmemory/service.go @@ -75,11 +75,12 @@ func (s *Service) Status(ctx context.Context, inst *store.ChannelInstanceData) ( if err != nil { return nil, err } - pending := 0 - for _, item := range items { - if item.Status == store.ChannelMemoryItemPendingReview { - pending++ - } + pending, err := s.Extractions.CountItems(ctx, store.ChannelMemoryItemListOptions{ + ChannelInstanceID: inst.ID, + Status: store.ChannelMemoryItemPendingReview, + }) + if err != nil { + return nil, err } unprocessed, err := s.UnprocessedMessageCount(ctx, inst) if err != nil { @@ -295,7 +296,7 @@ func (s *Service) unprocessedMessages(ctx context.Context, instID uuid.UUID, gro return nil, nil } -func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceData, cfg Config, group store.PendingMessageGroup, messages []store.PendingMessage, trigger string) (*store.ChannelMemoryExtractionRun, error) { +func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceData, cfg Config, group store.PendingMessageGroup, messages []store.PendingMessage, trigger string) (run *store.ChannelMemoryExtractionRun, err error) { if len(messages) < cfg.MinMessages { return nil, fmt.Errorf("not enough useful messages") } @@ -307,9 +308,13 @@ func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceDa if len(redacted.Messages) < cfg.MinMessages { return nil, fmt.Errorf("not enough redacted messages") } - start, end := redacted.Messages[0], redacted.Messages[len(redacted.Messages)-1] + consumed := messagesWithinExtractionBudget(redacted.Messages, extractionRetryMaxInputChars) + if len(consumed) < cfg.MinMessages { + return nil, fmt.Errorf("not enough extractable messages") + } + start, end := consumed[0], consumed[len(consumed)-1] redactionTypes, _ := json.Marshal(redacted.Types) - run := &store.ChannelMemoryExtractionRun{ + run = &store.ChannelMemoryExtractionRun{ ChannelInstanceID: inst.ID, ChannelName: inst.Name, AgentID: inst.AgentID, @@ -321,7 +326,7 @@ func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceDa SourceEndID: messageSourceID(end), SourceStartAt: &start.CreatedAt, SourceEndAt: &end.CreatedAt, - MessageCount: len(redacted.Messages), + MessageCount: len(consumed), RedactionCount: redacted.Count, RedactionTypes: redactionTypes, StartedAt: timePtr(time.Now().UTC()), @@ -329,12 +334,23 @@ func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceDa if err := s.Extractions.CreateRun(ctx, run); err != nil { return nil, err } - provider, model := providerresolve.ResolveBackgroundProvider(ctx, run.TenantID, s.Registry, s.SystemConfigs) - items, err := Extract(ctx, provider, model, s.UsageCaps, redacted.Messages, cfg.AllowedTypes) - if err != nil { + completed := false + defer func() { + if err == nil || completed { + return + } _ = s.Extractions.UpdateRun(ctx, run.ID, map[string]any{ - "status": store.ChannelMemoryRunFailed, "error_message": err.Error(), "completed_at": time.Now().UTC(), + "status": store.ChannelMemoryRunFailed, + "error_message": err.Error(), + "item_count": run.ItemCount, + "completed_at": time.Now().UTC(), }) + run.Status = store.ChannelMemoryRunFailed + run.ErrorMessage = err.Error() + }() + provider, model := providerresolve.ResolveBackgroundProvider(ctx, run.TenantID, s.Registry, s.SystemConfigs) + items, err := Extract(ctx, provider, model, s.UsageCaps, consumed, cfg.AllowedTypes) + if err != nil { return run, err } for _, extracted := range items { @@ -357,5 +373,6 @@ func (s *Service) runMessages(ctx context.Context, inst *store.ChannelInstanceDa "status": status, "item_count": run.ItemCount, "completed_at": time.Now().UTC(), }) run.Status = status + completed = true return run, nil } diff --git a/internal/channelmemory/service_test.go b/internal/channelmemory/service_test.go index 4a7e3b49..cdb442d4 100644 --- a/internal/channelmemory/service_test.go +++ b/internal/channelmemory/service_test.go @@ -3,18 +3,38 @@ package channelmemory import ( "context" "database/sql" + "errors" + "strings" "testing" "time" "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/store" ) type fakeExtractionStore struct { - runs []store.ChannelMemoryExtractionRun + runs []store.ChannelMemoryExtractionRun + items map[uuid.UUID]*store.ChannelMemoryExtractionItem + createdRun *store.ChannelMemoryExtractionRun + createdItems []*store.ChannelMemoryExtractionItem + createItemErr error + countItems int + countOpts store.ChannelMemoryItemListOptions + updateRuns []map[string]any + updateItems []map[string]any } -func (f *fakeExtractionStore) CreateRun(context.Context, *store.ChannelMemoryExtractionRun) error { +func (f *fakeExtractionStore) CreateRun(ctx context.Context, run *store.ChannelMemoryExtractionRun) error { + if run.ID == uuid.Nil { + run.ID = uuid.New() + } + if run.TenantID == uuid.Nil { + run.TenantID = store.TenantIDFromContext(ctx) + } + copied := *run + f.createdRun = &copied + f.runs = append([]store.ChannelMemoryExtractionRun{copied}, f.runs...) return nil } @@ -26,15 +46,48 @@ func (f *fakeExtractionStore) ListRuns(context.Context, store.ChannelMemoryRunLi return f.runs, nil } -func (f *fakeExtractionStore) UpdateRun(context.Context, uuid.UUID, map[string]any) error { +func (f *fakeExtractionStore) UpdateRun(_ context.Context, id uuid.UUID, updates map[string]any) error { + f.updateRuns = append(f.updateRuns, updates) + if f.createdRun != nil && f.createdRun.ID == id { + if status, ok := updates["status"].(string); ok { + f.createdRun.Status = status + } + if errMsg, ok := updates["error_message"].(string); ok { + f.createdRun.ErrorMessage = errMsg + } + if itemCount, ok := updates["item_count"].(int); ok { + f.createdRun.ItemCount = itemCount + } + } return nil } -func (f *fakeExtractionStore) CreateItem(context.Context, *store.ChannelMemoryExtractionItem) error { +func (f *fakeExtractionStore) CreateItem(_ context.Context, argumentsItem *store.ChannelMemoryExtractionItem) error { + if f.createItemErr != nil { + return f.createItemErr + } + if f.items == nil { + f.items = make(map[uuid.UUID]*store.ChannelMemoryExtractionItem) + } + item := *argumentsItem + if item.ID == uuid.Nil { + item.ID = uuid.New() + argumentsItem.ID = item.ID + } + if item.Status == "" { + item.Status = store.ChannelMemoryItemPendingReview + argumentsItem.Status = item.Status + } + f.items[item.ID] = &item + f.createdItems = append(f.createdItems, &item) return nil } -func (f *fakeExtractionStore) GetItem(context.Context, uuid.UUID) (*store.ChannelMemoryExtractionItem, error) { +func (f *fakeExtractionStore) GetItem(_ context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionItem, error) { + if item, ok := f.items[id]; ok { + copied := *item + return &copied, nil + } return nil, sql.ErrNoRows } @@ -42,11 +95,21 @@ func (f *fakeExtractionStore) ListItems(context.Context, store.ChannelMemoryItem return nil, nil } -func (f *fakeExtractionStore) CountItems(context.Context, store.ChannelMemoryItemListOptions) (int, error) { - return 0, nil +func (f *fakeExtractionStore) CountItems(_ context.Context, opts store.ChannelMemoryItemListOptions) (int, error) { + f.countOpts = opts + return f.countItems, nil } -func (f *fakeExtractionStore) UpdateItem(context.Context, uuid.UUID, map[string]any) error { +func (f *fakeExtractionStore) UpdateItem(_ context.Context, id uuid.UUID, updates map[string]any) error { + f.updateItems = append(f.updateItems, updates) + if item, ok := f.items[id]; ok { + if status, ok := updates["status"].(string); ok { + item.Status = status + } + if episodicID, ok := updates["episodic_id"].(string); ok { + item.EpisodicID = episodicID + } + } return nil } @@ -91,6 +154,53 @@ func (f *fakePendingStore) ResolveGroupTitles(context.Context, []store.PendingMe return nil, nil } +type fakeChannelStore struct { + store.ChannelInstanceStore + inst *store.ChannelInstanceData +} + +func (f *fakeChannelStore) Get(context.Context, uuid.UUID) (*store.ChannelInstanceData, error) { + if f.inst == nil { + return nil, sql.ErrNoRows + } + return f.inst, nil +} + +type fakeEpisodicStore struct { + store.EpisodicStore + exists bool + bySource *store.EpisodicSummary + created []*store.EpisodicSummary + getCalls int + createErr error + existsErr error + getByErr error +} + +func (f *fakeEpisodicStore) Create(_ context.Context, ep *store.EpisodicSummary) error { + if f.createErr != nil { + return f.createErr + } + if ep.ID == uuid.Nil { + ep.ID = uuid.New() + } + copied := *ep + f.created = append(f.created, &copied) + return nil +} + +func (f *fakeEpisodicStore) ExistsBySourceID(context.Context, string, string, string) (bool, error) { + return f.exists, f.existsErr +} + +func (f *fakeEpisodicStore) GetBySourceID(context.Context, string, string, string) (*store.EpisodicSummary, error) { + f.getCalls++ + if f.getByErr != nil { + return nil, f.getByErr + } + return f.bySource, nil +} + func TestShouldRunScheduledWhenMessageCapReached(t *testing.T) { cfg := DefaultConfig() cfg.MessageCap = 10 @@ -200,3 +310,205 @@ func TestItemHashIsStableAcrossRuns(t *testing.T) { t.Fatalf("hash changed across runs: %s != %s", itemA.ItemHash, itemB.ItemHash) } } + +func TestStatusCountsAllPendingItems(t *testing.T) { + inst := &store.ChannelInstanceData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + Name: "discord", + Config: MergeIntoInstanceConfig(nil, DefaultConfig()), + } + extractions := &fakeExtractionStore{countItems: 75} + svc := &Service{ + Pending: &fakePendingStore{}, + Extractions: extractions, + } + + status, err := svc.Status(context.Background(), inst) + if err != nil { + t.Fatal(err) + } + if status.PendingCount != 75 { + t.Fatalf("pending count = %d, want 75", status.PendingCount) + } + if extractions.countOpts.Status != store.ChannelMemoryItemPendingReview { + t.Fatalf("CountItems status = %q, want pending_review", extractions.countOpts.Status) + } + if extractions.countOpts.ChannelInstanceID != inst.ID { + t.Fatalf("CountItems channel = %s, want %s", extractions.countOpts.ChannelInstanceID, inst.ID) + } +} + +func TestRunMessagesCheckpointsOnlyExtractedBudget(t *testing.T) { + tenantID := uuid.New() + inst := &store.ChannelInstanceData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: tenantID, + Name: "discord", + AgentID: uuid.New(), + CreatedBy: "user-1", + } + cfg := DefaultConfig() + cfg.MinMessages = 2 + cfg.AllowedTypes = DefaultAllowedTypes + cfg.ReviewMode = true + messages := make([]store.PendingMessage, 0, 12) + for i := range 12 { + messages = append(messages, store.PendingMessage{ + ID: uuid.New(), + ChannelName: "discord", + HistoryKey: "group-a", + Sender: "tester", + Body: "This is durable project context. " + strings.Repeat("x", 900), + CreatedAt: time.Date(2026, 7, 7, 10, i, 0, 0, time.UTC), + }) + } + consumed := messagesWithinExtractionBudget(messages, extractionRetryMaxInputChars) + if len(consumed) == 0 || len(consumed) == len(messages) { + t.Fatalf("test fixture should be partially consumed, got %d of %d", len(consumed), len(messages)) + } + provider := &fakeExtractionProvider{responses: []providers.ChatResponse{{Content: `[]`, FinishReason: "stop"}}} + registry := providers.NewRegistry(store.TenantIDFromContext) + registry.RegisterForTenant(tenantID, provider) + extractions := &fakeExtractionStore{} + svc := &Service{Extractions: extractions, Registry: registry} + + ctx := store.WithTenantID(context.Background(), tenantID) + run, err := svc.runMessages(ctx, inst, cfg, store.PendingMessageGroup{ChannelName: "discord", HistoryKey: "group-a"}, messages, "manual") + if err != nil { + t.Fatal(err) + } + if run.MessageCount != len(consumed) { + t.Fatalf("run message_count = %d, want %d", run.MessageCount, len(consumed)) + } + wantEndID := messageSourceID(consumed[len(consumed)-1]) + if run.SourceEndID != wantEndID { + t.Fatalf("source_end_id = %s, want %s", run.SourceEndID, wantEndID) + } + if run.SourceEndID == messageSourceID(messages[len(messages)-1]) { + t.Fatal("run checkpoint advanced to a message outside the extraction budget") + } +} + +func TestRunMessagesMarksRunFailedWhenItemWriteFails(t *testing.T) { + tenantID := uuid.New() + inst := &store.ChannelInstanceData{ + BaseModel: store.BaseModel{ID: uuid.New()}, + TenantID: tenantID, + Name: "discord", + AgentID: uuid.New(), + CreatedBy: "user-1", + } + cfg := DefaultConfig() + cfg.MinMessages = 2 + cfg.AllowedTypes = DefaultAllowedTypes + cfg.ReviewMode = true + provider := &fakeExtractionProvider{responses: []providers.ChatResponse{{ + Content: `[{"type":"todos","summary":"Follow up on launch checklist","topics":["launch"],"entities":["GoClaw"],"confidence":0.91}]`, + FinishReason: "stop", + }}} + registry := providers.NewRegistry(store.TenantIDFromContext) + registry.RegisterForTenant(tenantID, provider) + extractions := &fakeExtractionStore{createItemErr: errors.New("create item boom")} + svc := &Service{Extractions: extractions, Registry: registry} + + ctx := store.WithTenantID(context.Background(), tenantID) + _, err := svc.runMessages(ctx, inst, cfg, store.PendingMessageGroup{ChannelName: "discord", HistoryKey: "group-a"}, extractionTestMessages(3), "manual") + if err == nil { + t.Fatal("expected CreateItem error") + } + if len(extractions.updateRuns) == 0 { + t.Fatal("expected failed run update") + } + last := extractions.updateRuns[len(extractions.updateRuns)-1] + if last["status"] != store.ChannelMemoryRunFailed { + t.Fatalf("run status update = %v, want failed", last["status"]) + } + if last["error_message"] != "create item boom" { + t.Fatalf("error_message = %v", last["error_message"]) + } +} + +func TestApproveUsesConfiguredRetentionHours(t *testing.T) { + instID := uuid.New() + agentID := uuid.New() + itemID := uuid.New() + cfg := DefaultConfig() + cfg.RetentionHours = 2 + extractions := &fakeExtractionStore{items: map[uuid.UUID]*store.ChannelMemoryExtractionItem{ + itemID: { + ID: itemID, + TenantID: uuid.New(), + ChannelInstanceID: instID, + AgentID: agentID, + UserID: "user-1", + SourceID: "channel:item-1", + Status: store.ChannelMemoryItemPendingReview, + Summary: "Remember the deployment window.", + }, + }} + episodic := &fakeEpisodicStore{} + svc := &Service{ + Extractions: extractions, + Episodic: episodic, + Channels: &fakeChannelStore{inst: &store.ChannelInstanceData{ + BaseModel: store.BaseModel{ID: instID}, + Config: MergeIntoInstanceConfig(nil, cfg), + }}, + } + + before := time.Now().UTC() + if _, err := svc.Approve(context.Background(), itemID, "admin"); err != nil { + t.Fatal(err) + } + if len(episodic.created) != 1 { + t.Fatalf("created episodic count = %d, want 1", len(episodic.created)) + } + expires := episodic.created[0].ExpiresAt + if expires == nil { + t.Fatal("episodic ExpiresAt is nil") + } + min := before.Add(2*time.Hour - time.Second) + max := time.Now().UTC().Add(2*time.Hour + time.Second) + if expires.Before(min) || expires.After(max) { + t.Fatalf("expires_at = %s, want around 2h from approval", expires) + } +} + +func TestApproveExistingSourceWritesExistingEpisodicID(t *testing.T) { + itemID := uuid.New() + existingID := uuid.New() + extractions := &fakeExtractionStore{items: map[uuid.UUID]*store.ChannelMemoryExtractionItem{ + itemID: { + ID: itemID, + TenantID: uuid.New(), + ChannelInstanceID: uuid.New(), + AgentID: uuid.New(), + UserID: "user-1", + SourceID: "channel:item-1", + Status: store.ChannelMemoryItemPendingReview, + Summary: "Remember the duplicate item.", + }, + }} + episodic := &fakeEpisodicStore{ + exists: true, + bySource: &store.EpisodicSummary{ID: existingID, SourceID: "channel:item-1"}, + } + svc := &Service{Extractions: extractions, Episodic: episodic} + + item, err := svc.Approve(context.Background(), itemID, "admin") + if err != nil { + t.Fatal(err) + } + if len(episodic.created) != 0 { + t.Fatalf("created episodic count = %d, want 0", len(episodic.created)) + } + if episodic.getCalls != 1 { + t.Fatalf("GetBySourceID calls = %d, want 1", episodic.getCalls) + } + if item.EpisodicID != existingID.String() { + t.Fatalf("item episodic_id = %q, want %q", item.EpisodicID, existingID) + } + if len(extractions.updateItems) == 0 || extractions.updateItems[len(extractions.updateItems)-1]["episodic_id"] != existingID.String() { + t.Fatalf("UpdateItem did not persist existing episodic_id: %#v", extractions.updateItems) + } +} diff --git a/internal/consolidation/workers_test.go b/internal/consolidation/workers_test.go index 42b7d835..ab6dda33 100644 --- a/internal/consolidation/workers_test.go +++ b/internal/consolidation/workers_test.go @@ -65,6 +65,15 @@ func (m *mockEpisodicStore) ExistsBySourceID(_ context.Context, _, _, sourceID s return m.existsByID[sourceID], nil } +func (m *mockEpisodicStore) GetBySourceID(_ context.Context, _, _, sourceID string) (*store.EpisodicSummary, error) { + m.mu.Lock() + defer m.mu.Unlock() + if !m.existsByID[sourceID] { + return nil, nil + } + return &store.EpisodicSummary{ID: uuid.New(), SourceID: sourceID}, nil +} + func (m *mockEpisodicStore) PruneExpired(_ context.Context) (int, error) { m.mu.Lock() defer m.mu.Unlock() diff --git a/internal/http/channel_memory_extraction.go b/internal/http/channel_memory_extraction.go index 0b03993b..7f228d8b 100644 --- a/internal/http/channel_memory_extraction.go +++ b/internal/http/channel_memory_extraction.go @@ -2,6 +2,7 @@ package http import ( "encoding/json" + "io" "net/http" "strconv" "strings" @@ -66,13 +67,18 @@ func (h *ChannelInstancesHandler) handleMemoryExtractionSettings(w http.Response if inst == nil { return } - var cfg channelmemory.Config - if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&cfg); err != nil { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20)) + if err != nil { + locale := store.LocaleFromContext(r.Context()) + writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)) + return + } + normalized, err := channelmemory.ApplyConfigPatch(channelmemory.ParseConfig(inst.Config), body) + if err != nil { locale := store.LocaleFromContext(r.Context()) writeError(w, http.StatusBadRequest, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)) return } - normalized := channelmemory.ParseConfig(channelmemory.MergeIntoInstanceConfig(nil, cfg)) configJSON := channelmemory.MergeIntoInstanceConfig(inst.Config, normalized) if err := h.store.Update(r.Context(), inst.ID, map[string]any{"config": configJSON}); err != nil { writeError(w, http.StatusInternalServerError, protocol.ErrInternal, "failed to update memory extraction settings") diff --git a/internal/store/episodic_store.go b/internal/store/episodic_store.go index ce390cc6..b60b7886 100644 --- a/internal/store/episodic_store.go +++ b/internal/store/episodic_store.go @@ -28,7 +28,7 @@ type EpisodicSummary struct { // Phase 10 — dreaming weighted scoring signals. Populated by // EpisodicStore.RecordRecall; consumed by consolidation.ComputeRecallScore. RecallCount int `json:"recall_count" db:"recall_count"` - RecallScore float64 `json:"recall_score" db:"recall_score"` // running average of memory_search hit scores + RecallScore float64 `json:"recall_score" db:"recall_score"` // running average of memory_search hit scores LastRecalledAt *time.Time `json:"last_recalled_at,omitempty" db:"last_recalled_at"` } @@ -64,6 +64,7 @@ type EpisodicStore interface { // Lifecycle ExistsBySourceID(ctx context.Context, agentID, userID, sourceID string) (bool, error) + GetBySourceID(ctx context.Context, agentID, userID, sourceID string) (*EpisodicSummary, error) PruneExpired(ctx context.Context) (int, error) // Promotion lifecycle (used by consolidation pipeline) diff --git a/internal/store/pg/channel_memory_extraction.go b/internal/store/pg/channel_memory_extraction.go index 24f6f0ef..61aa51f1 100644 --- a/internal/store/pg/channel_memory_extraction.go +++ b/internal/store/pg/channel_memory_extraction.go @@ -139,7 +139,7 @@ func (s *PGChannelMemoryExtractionStore) CreateItem(ctx context.Context, item *s err := s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_items (`+channelMemoryItemCols+`) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23) - ON CONFLICT (tenant_id, run_id, item_hash) DO UPDATE SET updated_at = EXCLUDED.updated_at + ON CONFLICT (tenant_id, channel_instance_id, item_hash) DO UPDATE SET updated_at = EXCLUDED.updated_at RETURNING id`, item.ID, item.TenantID, item.RunID, item.ChannelInstanceID, item.AgentID, item.UserID, item.ItemHash, item.ItemType, item.Summary, item.Topics, item.Entities, item.Confidence, diff --git a/internal/store/pg/episodic_summaries.go b/internal/store/pg/episodic_summaries.go index 77b3245e..307b9a38 100644 --- a/internal/store/pg/episodic_summaries.go +++ b/internal/store/pg/episodic_summaries.go @@ -25,7 +25,7 @@ func NewPGEpisodicStore(db *sql.DB) *PGEpisodicStore { } func (s *PGEpisodicStore) SetEmbeddingProvider(p store.EmbeddingProvider) { s.embProvider = p } -func (s *PGEpisodicStore) Close() error { return nil } +func (s *PGEpisodicStore) Close() error { return nil } // Create inserts a new episodic summary with optional embedding. func (s *PGEpisodicStore) Create(ctx context.Context, ep *store.EpisodicSummary) error { @@ -180,6 +180,17 @@ func (s *PGEpisodicStore) ExistsBySourceID(ctx context.Context, agentID, userID, return exists, err } +func (s *PGEpisodicStore) GetBySourceID(ctx context.Context, agentID, userID, sourceID string) (*store.EpisodicSummary, error) { + row := s.db.QueryRowContext(ctx, ` + SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, + turn_count, token_count, l0_abstract, source_id, source_type, + created_at, expires_at, recall_count, recall_score, last_recalled_at + FROM episodic_summaries + WHERE agent_id = $1 AND user_id = $2 AND source_id = $3 AND tenant_id = $4`, + agentID, userID, sourceID, store.TenantIDFromContext(ctx)) + return scanEpisodic(row) +} + // PruneExpired deletes all episodic summaries past their expiry across all tenants. // This is a global maintenance operation and does not filter by tenant. func (s *PGEpisodicStore) PruneExpired(ctx context.Context) (int, error) { diff --git a/internal/store/sqlitestore/channel-memory-extraction.go b/internal/store/sqlitestore/channel-memory-extraction.go index f76711d6..52c0f503 100644 --- a/internal/store/sqlitestore/channel-memory-extraction.go +++ b/internal/store/sqlitestore/channel-memory-extraction.go @@ -136,7 +136,7 @@ func (s *SQLiteChannelMemoryExtractionStore) CreateItem(ctx context.Context, ite return s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_items (`+sqliteChannelMemoryItemCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - ON CONFLICT (tenant_id, run_id, item_hash) DO UPDATE SET updated_at = excluded.updated_at + ON CONFLICT (tenant_id, channel_instance_id, item_hash) DO UPDATE SET updated_at = excluded.updated_at RETURNING id`, item.ID, item.TenantID, item.RunID, item.ChannelInstanceID, item.AgentID, item.UserID, item.ItemHash, item.ItemType, item.Summary, string(item.Topics), string(item.Entities), diff --git a/internal/store/sqlitestore/channel_memory_extraction_test.go b/internal/store/sqlitestore/channel_memory_extraction_test.go new file mode 100644 index 00000000..8af406aa --- /dev/null +++ b/internal/store/sqlitestore/channel_memory_extraction_test.go @@ -0,0 +1,97 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestChannelMemoryCreateItemDedupesAcrossRuns(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + tenantID := uuid.New() + agentID := uuid.New() + channelID := uuid.New() + if _, err := db.Exec(`INSERT INTO tenants (id, name, slug, status) VALUES (?, 'T', 't-channel-memory', 'active')`, tenantID.String()); err != nil { + t.Fatalf("insert tenant: %v", err) + } + if _, err := db.Exec(`INSERT INTO agents (id, tenant_id, agent_key, display_name, owner_id, provider, model) + VALUES (?, ?, 'agent-channel-memory', 'Agent', 'owner', 'openai', 'gpt-4o')`, agentID.String(), tenantID.String()); err != nil { + t.Fatalf("insert agent: %v", err) + } + if _, err := db.Exec(`INSERT INTO channel_instances (id, tenant_id, name, channel_type, agent_id) + VALUES (?, ?, 'discord', 'discord', ?)`, channelID.String(), tenantID.String(), agentID.String()); err != nil { + t.Fatalf("insert channel instance: %v", err) + } + + ctx := store.WithTenantID(context.Background(), tenantID) + extractions := NewSQLiteChannelMemoryExtractionStore(db) + runA := &store.ChannelMemoryExtractionRun{ + ChannelInstanceID: channelID, + ChannelName: "discord", + AgentID: agentID, + UserID: "user-1", + HistoryKey: "group-a", + SourceStartID: "msg-1", + SourceEndID: "msg-2", + } + if err := extractions.CreateRun(ctx, runA); err != nil { + t.Fatalf("CreateRun A: %v", err) + } + runB := &store.ChannelMemoryExtractionRun{ + ChannelInstanceID: channelID, + ChannelName: "discord", + AgentID: agentID, + UserID: "user-1", + HistoryKey: "group-a", + SourceStartID: "msg-3", + SourceEndID: "msg-4", + } + if err := extractions.CreateRun(ctx, runB); err != nil { + t.Fatalf("CreateRun B: %v", err) + } + + itemA := &store.ChannelMemoryExtractionItem{ + RunID: runA.ID, + ChannelInstanceID: channelID, + AgentID: agentID, + UserID: "user-1", + ItemHash: "stable-hash", + ItemType: "todos", + Summary: "Follow up on rollout checklist.", + SourceID: "channel:stable-hash", + } + if err := extractions.CreateItem(ctx, itemA); err != nil { + t.Fatalf("CreateItem A: %v", err) + } + itemB := &store.ChannelMemoryExtractionItem{ + RunID: runB.ID, + ChannelInstanceID: channelID, + AgentID: agentID, + UserID: "user-1", + ItemHash: "stable-hash", + ItemType: "todos", + Summary: "Follow up on rollout checklist.", + SourceID: "channel:stable-hash", + } + if err := extractions.CreateItem(ctx, itemB); err != nil { + t.Fatalf("CreateItem B: %v", err) + } + if itemB.ID != itemA.ID { + t.Fatalf("duplicate item ID = %s, want existing %s", itemB.ID, itemA.ID) + } + var count int + if err := db.QueryRow(`SELECT COUNT(*) FROM channel_memory_extraction_items WHERE tenant_id = ? AND channel_instance_id = ? AND item_hash = ?`, + tenantID.String(), channelID.String(), "stable-hash").Scan(&count); err != nil { + t.Fatalf("count items: %v", err) + } + if count != 1 { + t.Fatalf("duplicate item count = %d, want 1", count) + } +} diff --git a/internal/store/sqlitestore/episodic.go b/internal/store/sqlitestore/episodic.go index 2f63222b..c08b497a 100644 --- a/internal/store/sqlitestore/episodic.go +++ b/internal/store/sqlitestore/episodic.go @@ -131,6 +131,18 @@ func (s *SQLiteEpisodicStore) ExistsBySourceID(ctx context.Context, agentID, use return exists, err } +func (s *SQLiteEpisodicStore) GetBySourceID(ctx context.Context, agentID, userID, sourceID string) (*store.EpisodicSummary, error) { + tenantID := tenantIDForInsert(ctx) + row := s.db.QueryRowContext(ctx, ` + SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, + turn_count, token_count, l0_abstract, source_id, source_type, + created_at, expires_at, recall_count, recall_score, last_recalled_at + FROM episodic_summaries + WHERE agent_id = ? AND user_id = ? AND source_id = ? AND tenant_id = ?`, + agentID, userID, sourceID, tenantID.String()) + return scanSQLiteEpisodic(row) +} + // PruneExpired deletes all episodic summaries past their expiry across all tenants. // This is a global maintenance operation and does not filter by tenant. func (s *SQLiteEpisodicStore) PruneExpired(ctx context.Context) (int, error) { diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index b5e9f237..9ccccdf8 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -16,7 +16,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 53 +const SchemaVersion = 54 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -891,6 +891,8 @@ ALTER TABLE usage_events ADD COLUMN thinking_tokens BIGINT NOT NULL DEFAULT 0; ALTER TABLE usage_event_rollups ADD COLUMN cache_read_tokens BIGINT NOT NULL DEFAULT 0; ALTER TABLE usage_event_rollups ADD COLUMN cache_create_tokens BIGINT NOT NULL DEFAULT 0; ALTER TABLE usage_event_rollups ADD COLUMN thinking_tokens BIGINT NOT NULL DEFAULT 0;`, + // Version 53 → 54: dedupe passive memory extraction items across runs for the same channel instance. + 53: addChannelMemoryItemChannelHashUnique, } const addUsageEventAnalyticsTables = ` @@ -1128,13 +1130,40 @@ CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( episodic_id VARCHAR(64) NOT NULL DEFAULT '', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), - UNIQUE (tenant_id, run_id, item_hash) + UNIQUE (tenant_id, channel_instance_id, item_hash) ); CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status ON channel_memory_extraction_items(tenant_id, channel_instance_id, status, created_at DESC); CREATE INDEX IF NOT EXISTS idx_channel_memory_items_run ON channel_memory_extraction_items(tenant_id, run_id);` +const addChannelMemoryItemChannelHashUnique = ` +DELETE FROM channel_memory_extraction_items +WHERE id NOT IN ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER ( + PARTITION BY tenant_id, channel_instance_id, item_hash + ORDER BY + CASE status + WHEN 'written' THEN 5 + WHEN 'approved' THEN 4 + WHEN 'pending_review' THEN 3 + WHEN 'rejected' THEN 2 + WHEN 'deleted' THEN 1 + ELSE 0 + END DESC, + created_at DESC, + id DESC + ) AS rn + FROM channel_memory_extraction_items + ) + WHERE rn = 1 +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_channel_memory_items_tenant_channel_hash_unique + ON channel_memory_extraction_items(tenant_id, channel_instance_id, item_hash);` + const addChannelContextCapabilityTables = ` CREATE TABLE IF NOT EXISTS mcp_context_grants ( id TEXT NOT NULL PRIMARY KEY, diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index e6ffbaa4..a89859d1 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -1242,7 +1242,7 @@ CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( episodic_id VARCHAR(64) NOT NULL DEFAULT '', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), - UNIQUE (tenant_id, run_id, item_hash) + UNIQUE (tenant_id, channel_instance_id, item_hash) ); CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index aa4dbc98..4258f088 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 89 +const RequiredSchemaVersion uint = 90 diff --git a/migrations/000076_channel_memory_extraction.up.sql b/migrations/000076_channel_memory_extraction.up.sql index e454b8a3..804d2cc6 100644 --- a/migrations/000076_channel_memory_extraction.up.sql +++ b/migrations/000076_channel_memory_extraction.up.sql @@ -53,7 +53,7 @@ CREATE TABLE IF NOT EXISTS channel_memory_extraction_items ( episodic_id VARCHAR(64) NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - UNIQUE (tenant_id, run_id, item_hash) + UNIQUE (tenant_id, channel_instance_id, item_hash) ); CREATE INDEX IF NOT EXISTS idx_channel_memory_items_channel_status diff --git a/migrations/000090_channel_memory_item_dedupe.down.sql b/migrations/000090_channel_memory_item_dedupe.down.sql new file mode 100644 index 00000000..8469e1e8 --- /dev/null +++ b/migrations/000090_channel_memory_item_dedupe.down.sql @@ -0,0 +1,6 @@ +ALTER TABLE channel_memory_extraction_items + DROP CONSTRAINT IF EXISTS channel_memory_extraction_items_tenant_channel_hash_key; + +ALTER TABLE channel_memory_extraction_items + ADD CONSTRAINT channel_memory_extraction_items_tenant_id_run_id_item_hash_key + UNIQUE (tenant_id, run_id, item_hash); diff --git a/migrations/000090_channel_memory_item_dedupe.up.sql b/migrations/000090_channel_memory_item_dedupe.up.sql new file mode 100644 index 00000000..d43ae3b6 --- /dev/null +++ b/migrations/000090_channel_memory_item_dedupe.up.sql @@ -0,0 +1,36 @@ +DELETE FROM channel_memory_extraction_items i +USING channel_memory_extraction_items keep +WHERE i.tenant_id = keep.tenant_id + AND i.channel_instance_id = keep.channel_instance_id + AND i.item_hash = keep.item_hash + AND i.id <> keep.id + AND ( + CASE i.status + WHEN 'written' THEN 5 + WHEN 'approved' THEN 4 + WHEN 'pending_review' THEN 3 + WHEN 'rejected' THEN 2 + WHEN 'deleted' THEN 1 + ELSE 0 + END, + i.created_at, + i.id::text + ) < ( + CASE keep.status + WHEN 'written' THEN 5 + WHEN 'approved' THEN 4 + WHEN 'pending_review' THEN 3 + WHEN 'rejected' THEN 2 + WHEN 'deleted' THEN 1 + ELSE 0 + END, + keep.created_at, + keep.id::text + ); + +ALTER TABLE channel_memory_extraction_items + DROP CONSTRAINT IF EXISTS channel_memory_extraction_items_tenant_id_run_id_item_hash_key; + +ALTER TABLE channel_memory_extraction_items + ADD CONSTRAINT channel_memory_extraction_items_tenant_channel_hash_key + UNIQUE (tenant_id, channel_instance_id, item_hash);