fix(channelmemory): harden passive extraction reliability (#1386)

Co-authored-by: Collective Developer <man@collective.dev>

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
 

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

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 {

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")
+	}
+}

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, ", ") + `.

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()})
 }

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
 }

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)
+	}
+}

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()

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")

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)

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,

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) {

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),

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)
+	}
+}

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) {

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,

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

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

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

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);

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);