feat(channel-memory): add extraction store schema

internal/store/channel_memory_extraction_store.go

@@ -0,0 +1,99 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+const (
+	ChannelMemoryRunPending   = "pending"
+	ChannelMemoryRunRunning   = "running"
+	ChannelMemoryRunCompleted = "completed"
+	ChannelMemoryRunFailed    = "failed"
+
+	ChannelMemoryItemPendingReview = "pending_review"
+	ChannelMemoryItemApproved      = "approved"
+	ChannelMemoryItemRejected      = "rejected"
+	ChannelMemoryItemWritten       = "written"
+	ChannelMemoryItemDeleted       = "deleted"
+)
+
+type ChannelMemoryExtractionRun struct {
+	ID                uuid.UUID       `json:"id" db:"id"`
+	TenantID          uuid.UUID       `json:"tenant_id" db:"tenant_id"`
+	ChannelInstanceID uuid.UUID       `json:"channel_instance_id" db:"channel_instance_id"`
+	ChannelName       string          `json:"channel_name" db:"channel_name"`
+	AgentID           uuid.UUID       `json:"agent_id" db:"agent_id"`
+	UserID            string          `json:"user_id" db:"user_id"`
+	HistoryKey        string          `json:"history_key" db:"history_key"`
+	Trigger           string          `json:"trigger" db:"trigger"`
+	Status            string          `json:"status" db:"status"`
+	SourceStartID     string          `json:"source_start_id" db:"source_start_id"`
+	SourceEndID       string          `json:"source_end_id" db:"source_end_id"`
+	SourceStartAt     *time.Time      `json:"source_start_at,omitempty" db:"source_start_at"`
+	SourceEndAt       *time.Time      `json:"source_end_at,omitempty" db:"source_end_at"`
+	MessageCount      int             `json:"message_count" db:"message_count"`
+	RedactionCount    int             `json:"redaction_count" db:"redaction_count"`
+	RedactionTypes    json.RawMessage `json:"redaction_types" db:"redaction_types"`
+	ItemCount         int             `json:"item_count" db:"item_count"`
+	ErrorMessage      string          `json:"error_message,omitempty" db:"error_message"`
+	StartedAt         *time.Time      `json:"started_at,omitempty" db:"started_at"`
+	CompletedAt       *time.Time      `json:"completed_at,omitempty" db:"completed_at"`
+	CreatedAt         time.Time       `json:"created_at" db:"created_at"`
+	UpdatedAt         time.Time       `json:"updated_at" db:"updated_at"`
+}
+
+type ChannelMemoryExtractionItem struct {
+	ID                uuid.UUID       `json:"id" db:"id"`
+	TenantID          uuid.UUID       `json:"tenant_id" db:"tenant_id"`
+	RunID             uuid.UUID       `json:"run_id" db:"run_id"`
+	ChannelInstanceID uuid.UUID       `json:"channel_instance_id" db:"channel_instance_id"`
+	AgentID           uuid.UUID       `json:"agent_id" db:"agent_id"`
+	UserID            string          `json:"user_id" db:"user_id"`
+	ItemHash          string          `json:"item_hash" db:"item_hash"`
+	ItemType          string          `json:"item_type" db:"item_type"`
+	Summary           string          `json:"summary" db:"summary"`
+	Topics            json.RawMessage `json:"topics" db:"topics"`
+	Entities          json.RawMessage `json:"entities" db:"entities"`
+	Confidence        float64         `json:"confidence" db:"confidence"`
+	SourceID          string          `json:"source_id" db:"source_id"`
+	Status            string          `json:"status" db:"status"`
+	ApprovedBy        string          `json:"approved_by,omitempty" db:"approved_by"`
+	ApprovedAt        *time.Time      `json:"approved_at,omitempty" db:"approved_at"`
+	RejectedBy        string          `json:"rejected_by,omitempty" db:"rejected_by"`
+	RejectedAt        *time.Time      `json:"rejected_at,omitempty" db:"rejected_at"`
+	DeletedAt         *time.Time      `json:"deleted_at,omitempty" db:"deleted_at"`
+	WrittenAt         *time.Time      `json:"written_at,omitempty" db:"written_at"`
+	EpisodicID        string          `json:"episodic_id,omitempty" db:"episodic_id"`
+	CreatedAt         time.Time       `json:"created_at" db:"created_at"`
+	UpdatedAt         time.Time       `json:"updated_at" db:"updated_at"`
+}
+
+type ChannelMemoryRunListOptions struct {
+	ChannelInstanceID uuid.UUID
+	HistoryKey        string
+	Status            string
+	Limit             int
+}
+
+type ChannelMemoryItemListOptions struct {
+	ChannelInstanceID uuid.UUID
+	RunID             uuid.UUID
+	Status            string
+	Limit             int
+}
+
+type ChannelMemoryExtractionStore interface {
+	CreateRun(ctx context.Context, run *ChannelMemoryExtractionRun) error
+	GetRun(ctx context.Context, id uuid.UUID) (*ChannelMemoryExtractionRun, error)
+	ListRuns(ctx context.Context, opts ChannelMemoryRunListOptions) ([]ChannelMemoryExtractionRun, error)
+	UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error
+
+	CreateItem(ctx context.Context, item *ChannelMemoryExtractionItem) error
+	GetItem(ctx context.Context, id uuid.UUID) (*ChannelMemoryExtractionItem, error)
+	ListItems(ctx context.Context, opts ChannelMemoryItemListOptions) ([]ChannelMemoryExtractionItem, error)
+	UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error
+}

internal/store/pg/channel_memory_extraction.go

@@ -0,0 +1,239 @@
+package pg
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+)
+
+type PGChannelMemoryExtractionStore struct {
+	db *sql.DB
+}
+
+func NewPGChannelMemoryExtractionStore(db *sql.DB) *PGChannelMemoryExtractionStore {
+	return &PGChannelMemoryExtractionStore{db: db}
+}
+
+const channelMemoryRunCols = `id, tenant_id, channel_instance_id, channel_name, agent_id, user_id,
+ history_key, trigger, status, source_start_id, source_end_id, source_start_at, source_end_at,
+ message_count, redaction_count, redaction_types, item_count, error_message, started_at,
+ completed_at, created_at, updated_at`
+
+const channelMemoryItemCols = `id, tenant_id, run_id, channel_instance_id, agent_id, user_id,
+ item_hash, item_type, summary, topics, entities, confidence, source_id, status, approved_by,
+ approved_at, rejected_by, rejected_at, deleted_at, written_at, episodic_id, created_at, updated_at`
+
+func (s *PGChannelMemoryExtractionStore) CreateRun(ctx context.Context, run *store.ChannelMemoryExtractionRun) error {
+	if run.ID == uuid.Nil {
+		run.ID = store.GenNewID()
+	}
+	run.TenantID = tenantIDForInsert(ctx)
+	now := time.Now().UTC()
+	run.CreatedAt = now
+	run.UpdatedAt = now
+	if run.Status == "" {
+		run.Status = store.ChannelMemoryRunPending
+	}
+	if run.RedactionTypes == nil {
+		run.RedactionTypes = json.RawMessage("[]")
+	}
+	err := s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_runs
+		(`+channelMemoryRunCols+`)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)
+		ON CONFLICT (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id)
+		DO UPDATE SET updated_at = EXCLUDED.updated_at
+		RETURNING id`,
+		run.ID, run.TenantID, run.ChannelInstanceID, run.ChannelName, run.AgentID, run.UserID,
+		run.HistoryKey, run.Trigger, run.Status, run.SourceStartID, run.SourceEndID,
+		run.SourceStartAt, run.SourceEndAt, run.MessageCount, run.RedactionCount,
+		run.RedactionTypes, run.ItemCount, run.ErrorMessage, run.StartedAt, run.CompletedAt,
+		run.CreatedAt, run.UpdatedAt,
+	).Scan(&run.ID)
+	return err
+}
+
+func (s *PGChannelMemoryExtractionStore) GetRun(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionRun, error) {
+	tClause, tArgs, _, err := scopeClause(ctx, 2)
+	if err != nil {
+		return nil, err
+	}
+	row := s.db.QueryRowContext(ctx, `SELECT `+channelMemoryRunCols+` FROM channel_memory_extraction_runs WHERE id = $1`+tClause, append([]any{id}, tArgs...)...)
+	return scanChannelMemoryRun(row)
+}
+
+func (s *PGChannelMemoryExtractionStore) ListRuns(ctx context.Context, opts store.ChannelMemoryRunListOptions) ([]store.ChannelMemoryExtractionRun, error) {
+	where, args, next, err := scopeClause(ctx, 1)
+	if err != nil {
+		return nil, err
+	}
+	conds := []string{"1=1" + where}
+	if opts.ChannelInstanceID != uuid.Nil {
+		conds = append(conds, fmt.Sprintf("channel_instance_id = $%d", next))
+		args = append(args, opts.ChannelInstanceID)
+		next++
+	}
+	if opts.HistoryKey != "" {
+		conds = append(conds, fmt.Sprintf("history_key = $%d", next))
+		args = append(args, opts.HistoryKey)
+		next++
+	}
+	if opts.Status != "" {
+		conds = append(conds, fmt.Sprintf("status = $%d", next))
+		args = append(args, opts.Status)
+		next++
+	}
+	limit := opts.Limit
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	args = append(args, limit)
+	rows, err := s.db.QueryContext(ctx, `SELECT `+channelMemoryRunCols+`
+		FROM channel_memory_extraction_runs WHERE `+strings.Join(conds, " AND ")+`
+		ORDER BY created_at DESC LIMIT $`+fmt.Sprint(next), args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []store.ChannelMemoryExtractionRun
+	for rows.Next() {
+		run, err := scanChannelMemoryRunRows(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *run)
+	}
+	return out, rows.Err()
+}
+
+func (s *PGChannelMemoryExtractionStore) UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error {
+	tid, err := requireTenantID(ctx)
+	if err != nil {
+		return err
+	}
+	return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_runs", updates, id, tid)
+}
+
+func (s *PGChannelMemoryExtractionStore) CreateItem(ctx context.Context, item *store.ChannelMemoryExtractionItem) error {
+	if item.ID == uuid.Nil {
+		item.ID = store.GenNewID()
+	}
+	item.TenantID = tenantIDForInsert(ctx)
+	now := time.Now().UTC()
+	item.CreatedAt = now
+	item.UpdatedAt = now
+	if item.Status == "" {
+		item.Status = store.ChannelMemoryItemPendingReview
+	}
+	if item.Topics == nil {
+		item.Topics = json.RawMessage("[]")
+	}
+	if item.Entities == nil {
+		item.Entities = json.RawMessage("[]")
+	}
+	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
+		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,
+		item.SourceID, item.Status, item.ApprovedBy, item.ApprovedAt, item.RejectedBy,
+		item.RejectedAt, item.DeletedAt, item.WrittenAt, item.EpisodicID, item.CreatedAt, item.UpdatedAt,
+	).Scan(&item.ID)
+	return err
+}
+
+func (s *PGChannelMemoryExtractionStore) GetItem(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionItem, error) {
+	tClause, tArgs, _, err := scopeClause(ctx, 2)
+	if err != nil {
+		return nil, err
+	}
+	row := s.db.QueryRowContext(ctx, `SELECT `+channelMemoryItemCols+` FROM channel_memory_extraction_items WHERE id = $1`+tClause, append([]any{id}, tArgs...)...)
+	return scanChannelMemoryItem(row)
+}
+
+func (s *PGChannelMemoryExtractionStore) ListItems(ctx context.Context, opts store.ChannelMemoryItemListOptions) ([]store.ChannelMemoryExtractionItem, error) {
+	where, args, next, err := scopeClause(ctx, 1)
+	if err != nil {
+		return nil, err
+	}
+	conds := []string{"1=1" + where}
+	if opts.ChannelInstanceID != uuid.Nil {
+		conds = append(conds, fmt.Sprintf("channel_instance_id = $%d", next))
+		args = append(args, opts.ChannelInstanceID)
+		next++
+	}
+	if opts.RunID != uuid.Nil {
+		conds = append(conds, fmt.Sprintf("run_id = $%d", next))
+		args = append(args, opts.RunID)
+		next++
+	}
+	if opts.Status != "" {
+		conds = append(conds, fmt.Sprintf("status = $%d", next))
+		args = append(args, opts.Status)
+		next++
+	}
+	limit := opts.Limit
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	args = append(args, limit)
+	rows, err := s.db.QueryContext(ctx, `SELECT `+channelMemoryItemCols+`
+		FROM channel_memory_extraction_items WHERE `+strings.Join(conds, " AND ")+`
+		ORDER BY created_at DESC LIMIT $`+fmt.Sprint(next), args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []store.ChannelMemoryExtractionItem
+	for rows.Next() {
+		item, err := scanChannelMemoryItemRows(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *item)
+	}
+	return out, rows.Err()
+}
+
+func (s *PGChannelMemoryExtractionStore) UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error {
+	tid, err := requireTenantID(ctx)
+	if err != nil {
+		return err
+	}
+	return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_items", updates, id, tid)
+}
+
+type channelMemoryScanner interface {
+	Scan(dest ...any) error
+}
+
+func scanChannelMemoryRun(row channelMemoryScanner) (*store.ChannelMemoryExtractionRun, error) {
+	var r store.ChannelMemoryExtractionRun
+	if err := row.Scan(&r.ID, &r.TenantID, &r.ChannelInstanceID, &r.ChannelName, &r.AgentID, &r.UserID, &r.HistoryKey, &r.Trigger, &r.Status, &r.SourceStartID, &r.SourceEndID, &r.SourceStartAt, &r.SourceEndAt, &r.MessageCount, &r.RedactionCount, &r.RedactionTypes, &r.ItemCount, &r.ErrorMessage, &r.StartedAt, &r.CompletedAt, &r.CreatedAt, &r.UpdatedAt); err != nil {
+		return nil, err
+	}
+	return &r, nil
+}
+
+func scanChannelMemoryRunRows(rows *sql.Rows) (*store.ChannelMemoryExtractionRun, error) {
+	return scanChannelMemoryRun(rows)
+}
+
+func scanChannelMemoryItem(row channelMemoryScanner) (*store.ChannelMemoryExtractionItem, error) {
+	var i store.ChannelMemoryExtractionItem
+	if err := row.Scan(&i.ID, &i.TenantID, &i.RunID, &i.ChannelInstanceID, &i.AgentID, &i.UserID, &i.ItemHash, &i.ItemType, &i.Summary, &i.Topics, &i.Entities, &i.Confidence, &i.SourceID, &i.Status, &i.ApprovedBy, &i.ApprovedAt, &i.RejectedBy, &i.RejectedAt, &i.DeletedAt, &i.WrittenAt, &i.EpisodicID, &i.CreatedAt, &i.UpdatedAt); err != nil {
+		return nil, err
+	}
+	return &i, nil
+}
+
+func scanChannelMemoryItemRows(rows *sql.Rows) (*store.ChannelMemoryExtractionItem, error) {
+	return scanChannelMemoryItem(rows)
+}

internal/store/pg/factory.go

@@ -40,6 +40,7 @@ func NewPGStores(cfg store.StoreConfig) (*store.Stores, error) {
 		Teams:                  NewPGTeamStore(db),
 		BuiltinTools:           NewPGBuiltinToolStore(db),
 		PendingMessages:        NewPGPendingMessageStore(db),
+		ChannelMemory:          NewPGChannelMemoryExtractionStore(db),
 		KnowledgeGraph:         NewPGKnowledgeGraphStore(db),
 		Contacts:               NewPGContactStore(db),
 		Activity:               NewPGActivityStore(db),

internal/store/sqlitestore/channel-memory-extraction.go

@@ -0,0 +1,257 @@
+//go:build sqlite || sqliteonly
+
+package sqlitestore
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+)
+
+type SQLiteChannelMemoryExtractionStore struct {
+	db *sql.DB
+}
+
+func NewSQLiteChannelMemoryExtractionStore(db *sql.DB) *SQLiteChannelMemoryExtractionStore {
+	return &SQLiteChannelMemoryExtractionStore{db: db}
+}
+
+const sqliteChannelMemoryRunCols = `id, tenant_id, channel_instance_id, channel_name, agent_id, user_id,
+ history_key, trigger, status, source_start_id, source_end_id, source_start_at, source_end_at,
+ message_count, redaction_count, redaction_types, item_count, error_message, started_at,
+ completed_at, created_at, updated_at`
+
+const sqliteChannelMemoryItemCols = `id, tenant_id, run_id, channel_instance_id, agent_id, user_id,
+ item_hash, item_type, summary, topics, entities, confidence, source_id, status, approved_by,
+ approved_at, rejected_by, rejected_at, deleted_at, written_at, episodic_id, created_at, updated_at`
+
+func (s *SQLiteChannelMemoryExtractionStore) CreateRun(ctx context.Context, run *store.ChannelMemoryExtractionRun) error {
+	if run.ID == uuid.Nil {
+		run.ID = store.GenNewID()
+	}
+	run.TenantID = tenantIDForInsert(ctx)
+	now := time.Now().UTC()
+	run.CreatedAt = now
+	run.UpdatedAt = now
+	if run.Status == "" {
+		run.Status = store.ChannelMemoryRunPending
+	}
+	if run.RedactionTypes == nil {
+		run.RedactionTypes = json.RawMessage("[]")
+	}
+	return s.db.QueryRowContext(ctx, `INSERT INTO channel_memory_extraction_runs
+		(`+sqliteChannelMemoryRunCols+`)
+		VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+		ON CONFLICT (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id)
+		DO UPDATE SET updated_at = excluded.updated_at
+		RETURNING id`,
+		run.ID, run.TenantID, run.ChannelInstanceID, run.ChannelName, run.AgentID, run.UserID,
+		run.HistoryKey, run.Trigger, run.Status, run.SourceStartID, run.SourceEndID,
+		run.SourceStartAt, run.SourceEndAt, run.MessageCount, run.RedactionCount,
+		string(run.RedactionTypes), run.ItemCount, run.ErrorMessage, run.StartedAt, run.CompletedAt,
+		run.CreatedAt, run.UpdatedAt,
+	).Scan(&run.ID)
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) GetRun(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionRun, error) {
+	tClause, tArgs, err := scopeClause(ctx)
+	if err != nil {
+		return nil, err
+	}
+	row := s.db.QueryRowContext(ctx, `SELECT `+sqliteChannelMemoryRunCols+` FROM channel_memory_extraction_runs WHERE id = ?`+tClause, append([]any{id}, tArgs...)...)
+	return scanSQLiteChannelMemoryRun(row)
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) ListRuns(ctx context.Context, opts store.ChannelMemoryRunListOptions) ([]store.ChannelMemoryExtractionRun, error) {
+	tClause, args, err := scopeClause(ctx)
+	if err != nil {
+		return nil, err
+	}
+	conds := []string{"1=1" + tClause}
+	if opts.ChannelInstanceID != uuid.Nil {
+		conds = append(conds, "channel_instance_id = ?")
+		args = append(args, opts.ChannelInstanceID)
+	}
+	if opts.HistoryKey != "" {
+		conds = append(conds, "history_key = ?")
+		args = append(args, opts.HistoryKey)
+	}
+	if opts.Status != "" {
+		conds = append(conds, "status = ?")
+		args = append(args, opts.Status)
+	}
+	limit := opts.Limit
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	args = append(args, limit)
+	rows, err := s.db.QueryContext(ctx, `SELECT `+sqliteChannelMemoryRunCols+`
+		FROM channel_memory_extraction_runs WHERE `+strings.Join(conds, " AND ")+`
+		ORDER BY created_at DESC LIMIT ?`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []store.ChannelMemoryExtractionRun
+	for rows.Next() {
+		run, err := scanSQLiteChannelMemoryRun(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *run)
+	}
+	return out, rows.Err()
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) UpdateRun(ctx context.Context, id uuid.UUID, updates map[string]any) error {
+	tid, err := requireTenantID(ctx)
+	if err != nil {
+		return err
+	}
+	return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_runs", updates, id, tid)
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) CreateItem(ctx context.Context, item *store.ChannelMemoryExtractionItem) error {
+	if item.ID == uuid.Nil {
+		item.ID = store.GenNewID()
+	}
+	item.TenantID = tenantIDForInsert(ctx)
+	now := time.Now().UTC()
+	item.CreatedAt = now
+	item.UpdatedAt = now
+	if item.Status == "" {
+		item.Status = store.ChannelMemoryItemPendingReview
+	}
+	if item.Topics == nil {
+		item.Topics = json.RawMessage("[]")
+	}
+	if item.Entities == nil {
+		item.Entities = json.RawMessage("[]")
+	}
+	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
+		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),
+		item.Confidence, item.SourceID, item.Status, item.ApprovedBy, item.ApprovedAt,
+		item.RejectedBy, item.RejectedAt, item.DeletedAt, item.WrittenAt, item.EpisodicID,
+		item.CreatedAt, item.UpdatedAt,
+	).Scan(&item.ID)
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) GetItem(ctx context.Context, id uuid.UUID) (*store.ChannelMemoryExtractionItem, error) {
+	tClause, tArgs, err := scopeClause(ctx)
+	if err != nil {
+		return nil, err
+	}
+	row := s.db.QueryRowContext(ctx, `SELECT `+sqliteChannelMemoryItemCols+` FROM channel_memory_extraction_items WHERE id = ?`+tClause, append([]any{id}, tArgs...)...)
+	return scanSQLiteChannelMemoryItem(row)
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) ListItems(ctx context.Context, opts store.ChannelMemoryItemListOptions) ([]store.ChannelMemoryExtractionItem, error) {
+	tClause, args, err := scopeClause(ctx)
+	if err != nil {
+		return nil, err
+	}
+	conds := []string{"1=1" + tClause}
+	if opts.ChannelInstanceID != uuid.Nil {
+		conds = append(conds, "channel_instance_id = ?")
+		args = append(args, opts.ChannelInstanceID)
+	}
+	if opts.RunID != uuid.Nil {
+		conds = append(conds, "run_id = ?")
+		args = append(args, opts.RunID)
+	}
+	if opts.Status != "" {
+		conds = append(conds, "status = ?")
+		args = append(args, opts.Status)
+	}
+	limit := opts.Limit
+	if limit <= 0 || limit > 200 {
+		limit = 50
+	}
+	args = append(args, limit)
+	rows, err := s.db.QueryContext(ctx, `SELECT `+sqliteChannelMemoryItemCols+`
+		FROM channel_memory_extraction_items WHERE `+strings.Join(conds, " AND ")+`
+		ORDER BY created_at DESC LIMIT ?`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []store.ChannelMemoryExtractionItem
+	for rows.Next() {
+		item, err := scanSQLiteChannelMemoryItem(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *item)
+	}
+	return out, rows.Err()
+}
+
+func (s *SQLiteChannelMemoryExtractionStore) UpdateItem(ctx context.Context, id uuid.UUID, updates map[string]any) error {
+	tid, err := requireTenantID(ctx)
+	if err != nil {
+		return err
+	}
+	return execMapUpdateWhereTenant(ctx, s.db, "channel_memory_extraction_items", updates, id, tid)
+}
+
+type sqliteChannelMemoryScanner interface {
+	Scan(dest ...any) error
+}
+
+func scanSQLiteChannelMemoryRun(row sqliteChannelMemoryScanner) (*store.ChannelMemoryExtractionRun, error) {
+	var r store.ChannelMemoryExtractionRun
+	var redactions string
+	sourceStart, sourceEnd := &nullSqliteTime{}, &nullSqliteTime{}
+	started, completed := &nullSqliteTime{}, &nullSqliteTime{}
+	created, updated := scanTimePair()
+	if err := row.Scan(&r.ID, &r.TenantID, &r.ChannelInstanceID, &r.ChannelName, &r.AgentID, &r.UserID, &r.HistoryKey, &r.Trigger, &r.Status, &r.SourceStartID, &r.SourceEndID, sourceStart, sourceEnd, &r.MessageCount, &r.RedactionCount, &redactions, &r.ItemCount, &r.ErrorMessage, started, completed, created, updated); err != nil {
+		return nil, err
+	}
+	r.RedactionTypes = json.RawMessage(redactions)
+	r.SourceStartAt = sqliteTimePtr(sourceStart)
+	r.SourceEndAt = sqliteTimePtr(sourceEnd)
+	r.StartedAt = sqliteTimePtr(started)
+	r.CompletedAt = sqliteTimePtr(completed)
+	r.CreatedAt = created.Time
+	r.UpdatedAt = updated.Time
+	return &r, nil
+}
+
+func scanSQLiteChannelMemoryItem(row sqliteChannelMemoryScanner) (*store.ChannelMemoryExtractionItem, error) {
+	var i store.ChannelMemoryExtractionItem
+	var topics, entities string
+	approved, rejected := &nullSqliteTime{}, &nullSqliteTime{}
+	deleted, written := &nullSqliteTime{}, &nullSqliteTime{}
+	created, updated := scanTimePair()
+	if err := row.Scan(&i.ID, &i.TenantID, &i.RunID, &i.ChannelInstanceID, &i.AgentID, &i.UserID, &i.ItemHash, &i.ItemType, &i.Summary, &topics, &entities, &i.Confidence, &i.SourceID, &i.Status, &i.ApprovedBy, approved, &i.RejectedBy, rejected, deleted, written, &i.EpisodicID, created, updated); err != nil {
+		return nil, err
+	}
+	i.Topics = json.RawMessage(topics)
+	i.Entities = json.RawMessage(entities)
+	i.ApprovedAt = sqliteTimePtr(approved)
+	i.RejectedAt = sqliteTimePtr(rejected)
+	i.DeletedAt = sqliteTimePtr(deleted)
+	i.WrittenAt = sqliteTimePtr(written)
+	i.CreatedAt = created.Time
+	i.UpdatedAt = updated.Time
+	return &i, nil
+}
+
+func sqliteTimePtr(nt *nullSqliteTime) *time.Time {
+	if nt == nil || !nt.Valid {
+		return nil
+	}
+	return &nt.Time
+}
+
+var _ store.ChannelMemoryExtractionStore = (*SQLiteChannelMemoryExtractionStore)(nil)

internal/store/sqlitestore/factory.go

@@ -36,43 +36,44 @@ func NewSQLiteStores(cfg store.StoreConfig) (*store.Stores, error) {
 	}
 
 	sqliteStores := &store.Stores{
-		DB:                    db,
-		Sessions:              NewSQLiteSessionStore(db),
-		Agents:                NewSQLiteAgentStore(db),
-		Providers:             NewSQLiteProviderStore(db, cfg.EncryptionKey),
-		Tracing:               NewSQLiteTracingStore(db),
-		ConfigSecrets:         NewSQLiteConfigSecretsStore(db, cfg.EncryptionKey),
-		BuiltinTools:          NewSQLiteBuiltinToolStore(db),
-		Heartbeats:            NewSQLiteHeartbeatStore(db),
-		Tenants:               NewSQLiteTenantStore(db),
-		BuiltinToolTenantCfgs: NewSQLiteBuiltinToolTenantConfigStore(db),
-		SkillTenantCfgs:       NewSQLiteSkillTenantConfigStore(db),
-		SystemConfigs:         NewSQLiteSystemConfigStore(db),
-		Snapshots:             NewSQLiteSnapshotStore(db),
-		Cron:                  NewSQLiteCronStore(db),
-		ChannelInstances:      NewSQLiteChannelInstanceStore(db, cfg.EncryptionKey),
-		Pairing:               NewSQLitePairingStore(db),
-		PendingMessages:       NewSQLitePendingMessageStore(db),
-		Contacts:              NewSQLiteContactStore(db),
-		Teams:  NewSQLiteTeamStore(db),
-		Skills: NewSQLiteSkillStore(db, cfg.SkillsStorageDir),
-		MCP:    NewSQLiteMCPServerStore(db, cfg.EncryptionKey),
-		Activity:         NewSQLiteActivityStore(db),
-		APIKeys:          NewSQLiteAPIKeyStore(db),
-		ConfigPermissions: NewSQLiteConfigPermissionStore(db),
-		BrowserCookies:   NewSQLiteBrowserCookieStore(db, cfg.EncryptionKey),
-		Memory:         NewSQLiteMemoryStore(db),
-		SubagentTasks:   NewSQLiteSubagentTaskStore(db),
-		AgentLinks:      NewSQLiteAgentLinkStore(db),
-		SecureCLI:            secureCLI,
-		SecureCLIGrants:      NewSQLiteSecureCLIAgentGrantStore(db, cfg.EncryptionKey),
-		Episodic:             NewSQLiteEpisodicStore(db),
-		EvolutionMetrics:     NewSQLiteEvolutionMetricsStore(db),
-		EvolutionSuggestions: NewSQLiteEvolutionSuggestionStore(db),
-		KnowledgeGraph:       NewSQLiteKnowledgeGraphStore(db),
-		Vault:                NewSQLiteVaultStore(db),
-		BitrixPortals:        NewSQLiteBitrixPortalStore(db, cfg.EncryptionKey),
-		Hooks:                NewSQLiteHookStore(db),
+		DB:                     db,
+		Sessions:               NewSQLiteSessionStore(db),
+		Agents:                 NewSQLiteAgentStore(db),
+		Providers:              NewSQLiteProviderStore(db, cfg.EncryptionKey),
+		Tracing:                NewSQLiteTracingStore(db),
+		ConfigSecrets:          NewSQLiteConfigSecretsStore(db, cfg.EncryptionKey),
+		BuiltinTools:           NewSQLiteBuiltinToolStore(db),
+		Heartbeats:             NewSQLiteHeartbeatStore(db),
+		Tenants:                NewSQLiteTenantStore(db),
+		BuiltinToolTenantCfgs:  NewSQLiteBuiltinToolTenantConfigStore(db),
+		SkillTenantCfgs:        NewSQLiteSkillTenantConfigStore(db),
+		SystemConfigs:          NewSQLiteSystemConfigStore(db),
+		Snapshots:              NewSQLiteSnapshotStore(db),
+		Cron:                   NewSQLiteCronStore(db),
+		ChannelInstances:       NewSQLiteChannelInstanceStore(db, cfg.EncryptionKey),
+		Pairing:                NewSQLitePairingStore(db),
+		PendingMessages:        NewSQLitePendingMessageStore(db),
+		ChannelMemory:          NewSQLiteChannelMemoryExtractionStore(db),
+		Contacts:               NewSQLiteContactStore(db),
+		Teams:                  NewSQLiteTeamStore(db),
+		Skills:                 NewSQLiteSkillStore(db, cfg.SkillsStorageDir),
+		MCP:                    NewSQLiteMCPServerStore(db, cfg.EncryptionKey),
+		Activity:               NewSQLiteActivityStore(db),
+		APIKeys:                NewSQLiteAPIKeyStore(db),
+		ConfigPermissions:      NewSQLiteConfigPermissionStore(db),
+		BrowserCookies:         NewSQLiteBrowserCookieStore(db, cfg.EncryptionKey),
+		Memory:                 NewSQLiteMemoryStore(db),
+		SubagentTasks:          NewSQLiteSubagentTaskStore(db),
+		AgentLinks:             NewSQLiteAgentLinkStore(db),
+		SecureCLI:              secureCLI,
+		SecureCLIGrants:        NewSQLiteSecureCLIAgentGrantStore(db, cfg.EncryptionKey),
+		Episodic:               NewSQLiteEpisodicStore(db),
+		EvolutionMetrics:       NewSQLiteEvolutionMetricsStore(db),
+		EvolutionSuggestions:   NewSQLiteEvolutionSuggestionStore(db),
+		KnowledgeGraph:         NewSQLiteKnowledgeGraphStore(db),
+		Vault:                  NewSQLiteVaultStore(db),
+		BitrixPortals:          NewSQLiteBitrixPortalStore(db, cfg.EncryptionKey),
+		Hooks:                  NewSQLiteHookStore(db),
 		Webhooks:               NewSQLiteWebhookStore(db),
 		WebhookCalls:           NewSQLiteWebhookCallStore(db),
 		Workstations:           NewSQLiteWorkstationStore(db, cfg.EncryptionKey),

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 = 43
+const SchemaVersion = 44
 
 // migrations maps version → SQL to apply when upgrading FROM that version.
 // schema.sql always represents the LATEST full schema (for fresh DBs).
@@ -774,8 +774,72 @@ CREATE INDEX IF NOT EXISTS idx_browser_cookies_expires_at
 	41: `ALTER TABLE secure_cli_binaries ADD COLUMN adapter_name TEXT;`,
 	// Version 42 → 43: channel-context MCP and Secure CLI grants/credentials.
 	42: addChannelContextCapabilityTables,
+	// Version 43 → 44: passive channel memory extraction run and review queue.
+	43: addChannelMemoryExtractionTables,
 }
 
+const addChannelMemoryExtractionTables = `
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs (
+    id                  TEXT NOT NULL PRIMARY KEY,
+    tenant_id           TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    channel_name        VARCHAR(255) NOT NULL,
+    agent_id            TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id             VARCHAR(255) NOT NULL DEFAULT '',
+    history_key         VARCHAR(255) NOT NULL,
+    trigger             VARCHAR(32) NOT NULL DEFAULT 'scheduled',
+    status              VARCHAR(32) NOT NULL DEFAULT 'pending',
+    source_start_id     VARCHAR(255) NOT NULL DEFAULT '',
+    source_end_id       VARCHAR(255) NOT NULL DEFAULT '',
+    source_start_at     TEXT,
+    source_end_at       TEXT,
+    message_count       INTEGER NOT NULL DEFAULT 0,
+    redaction_count     INTEGER NOT NULL DEFAULT 0,
+    redaction_types     TEXT NOT NULL DEFAULT '[]',
+    item_count          INTEGER NOT NULL DEFAULT 0,
+    error_message       TEXT NOT NULL DEFAULT '',
+    started_at          TEXT,
+    completed_at        TEXT,
+    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, channel_instance_id, history_key, source_start_id, source_end_id)
+);
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel
+  ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status
+  ON channel_memory_extraction_runs(tenant_id, status, created_at DESC);
+
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_items (
+    id                  TEXT NOT NULL PRIMARY KEY,
+    tenant_id           TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    run_id              TEXT NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE,
+    channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    agent_id            TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id             VARCHAR(255) NOT NULL DEFAULT '',
+    item_hash           VARCHAR(128) NOT NULL,
+    item_type           VARCHAR(64) NOT NULL,
+    summary             TEXT NOT NULL,
+    topics              TEXT NOT NULL DEFAULT '[]',
+    entities            TEXT NOT NULL DEFAULT '[]',
+    confidence          REAL NOT NULL DEFAULT 0,
+    source_id           VARCHAR(255) NOT NULL DEFAULT '',
+    status              VARCHAR(32) NOT NULL DEFAULT 'pending_review',
+    approved_by         VARCHAR(255) NOT NULL DEFAULT '',
+    approved_at         TEXT,
+    rejected_by         VARCHAR(255) NOT NULL DEFAULT '',
+    rejected_at         TEXT,
+    deleted_at          TEXT,
+    written_at          TEXT,
+    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)
+);
+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 addChannelContextCapabilityTables = `
 CREATE TABLE IF NOT EXISTS mcp_context_grants (
     id                  TEXT NOT NULL PRIMARY KEY,

internal/store/sqlitestore/schema.sql

@@ -1141,6 +1141,77 @@ CREATE TABLE IF NOT EXISTS channel_pending_messages (
 CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_lookup ON channel_pending_messages(channel_name, history_key, created_at);
 CREATE INDEX IF NOT EXISTS idx_channel_pending_messages_tenant ON channel_pending_messages(tenant_id);
 
+-- ============================================================
+-- Table: channel_memory_extraction_runs
+-- ============================================================
+
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs (
+    id                  TEXT NOT NULL PRIMARY KEY,
+    tenant_id           TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    channel_name        VARCHAR(255) NOT NULL,
+    agent_id            TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id             VARCHAR(255) NOT NULL DEFAULT '',
+    history_key         VARCHAR(255) NOT NULL,
+    trigger             VARCHAR(32) NOT NULL DEFAULT 'scheduled',
+    status              VARCHAR(32) NOT NULL DEFAULT 'pending',
+    source_start_id     VARCHAR(255) NOT NULL DEFAULT '',
+    source_end_id       VARCHAR(255) NOT NULL DEFAULT '',
+    source_start_at     TEXT,
+    source_end_at       TEXT,
+    message_count       INTEGER NOT NULL DEFAULT 0,
+    redaction_count     INTEGER NOT NULL DEFAULT 0,
+    redaction_types     TEXT NOT NULL DEFAULT '[]',
+    item_count          INTEGER NOT NULL DEFAULT 0,
+    error_message       TEXT NOT NULL DEFAULT '',
+    started_at          TEXT,
+    completed_at        TEXT,
+    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, channel_instance_id, history_key, source_start_id, source_end_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel
+    ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status
+    ON channel_memory_extraction_runs(tenant_id, status, created_at DESC);
+
+-- ============================================================
+-- Table: channel_memory_extraction_items
+-- ============================================================
+
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_items (
+    id                  TEXT NOT NULL PRIMARY KEY,
+    tenant_id           TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    run_id              TEXT NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE,
+    channel_instance_id TEXT NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    agent_id            TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id             VARCHAR(255) NOT NULL DEFAULT '',
+    item_hash           VARCHAR(128) NOT NULL,
+    item_type           VARCHAR(64) NOT NULL,
+    summary             TEXT NOT NULL,
+    topics              TEXT NOT NULL DEFAULT '[]',
+    entities            TEXT NOT NULL DEFAULT '[]',
+    confidence          REAL NOT NULL DEFAULT 0,
+    source_id           VARCHAR(255) NOT NULL DEFAULT '',
+    status              VARCHAR(32) NOT NULL DEFAULT 'pending_review',
+    approved_by         VARCHAR(255) NOT NULL DEFAULT '',
+    approved_at         TEXT,
+    rejected_by         VARCHAR(255) NOT NULL DEFAULT '',
+    rejected_at         TEXT,
+    deleted_at          TEXT,
+    written_at          TEXT,
+    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)
+);
+
+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);
+
 -- ============================================================
 -- Table: channel_contacts
 -- ============================================================

internal/store/stores.go

@@ -20,6 +20,7 @@ type Stores struct {
 	Teams                 TeamStore
 	BuiltinTools          BuiltinToolStore
 	PendingMessages       PendingMessageStore
+	ChannelMemory         ChannelMemoryExtractionStore
 	KnowledgeGraph        KnowledgeGraphStore
 	Contacts              ContactStore
 	Activity              ActivityStore

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 = 74
+const RequiredSchemaVersion uint = 75

migrations/000075_channel_memory_extraction.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS channel_memory_extraction_items;
+DROP TABLE IF EXISTS channel_memory_extraction_runs;

migrations/000075_channel_memory_extraction.up.sql

@@ -0,0 +1,62 @@
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_runs (
+    id UUID PRIMARY KEY,
+    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    channel_instance_id UUID NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    channel_name VARCHAR(255) NOT NULL,
+    agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id VARCHAR(255) NOT NULL DEFAULT '',
+    history_key VARCHAR(255) NOT NULL,
+    trigger VARCHAR(32) NOT NULL DEFAULT 'scheduled',
+    status VARCHAR(32) NOT NULL DEFAULT 'pending',
+    source_start_id VARCHAR(255) NOT NULL DEFAULT '',
+    source_end_id VARCHAR(255) NOT NULL DEFAULT '',
+    source_start_at TIMESTAMPTZ,
+    source_end_at TIMESTAMPTZ,
+    message_count INTEGER NOT NULL DEFAULT 0,
+    redaction_count INTEGER NOT NULL DEFAULT 0,
+    redaction_types JSONB NOT NULL DEFAULT '[]'::jsonb,
+    item_count INTEGER NOT NULL DEFAULT 0,
+    error_message TEXT NOT NULL DEFAULT '',
+    started_at TIMESTAMPTZ,
+    completed_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE (tenant_id, channel_instance_id, history_key, source_start_id, source_end_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_channel
+    ON channel_memory_extraction_runs(tenant_id, channel_instance_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_channel_memory_runs_status
+    ON channel_memory_extraction_runs(tenant_id, status, created_at DESC);
+
+CREATE TABLE IF NOT EXISTS channel_memory_extraction_items (
+    id UUID PRIMARY KEY,
+    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
+    run_id UUID NOT NULL REFERENCES channel_memory_extraction_runs(id) ON DELETE CASCADE,
+    channel_instance_id UUID NOT NULL REFERENCES channel_instances(id) ON DELETE CASCADE,
+    agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
+    user_id VARCHAR(255) NOT NULL DEFAULT '',
+    item_hash VARCHAR(128) NOT NULL,
+    item_type VARCHAR(64) NOT NULL,
+    summary TEXT NOT NULL,
+    topics JSONB NOT NULL DEFAULT '[]'::jsonb,
+    entities JSONB NOT NULL DEFAULT '[]'::jsonb,
+    confidence DOUBLE PRECISION NOT NULL DEFAULT 0,
+    source_id VARCHAR(255) NOT NULL DEFAULT '',
+    status VARCHAR(32) NOT NULL DEFAULT 'pending_review',
+    approved_by VARCHAR(255) NOT NULL DEFAULT '',
+    approved_at TIMESTAMPTZ,
+    rejected_by VARCHAR(255) NOT NULL DEFAULT '',
+    rejected_at TIMESTAMPTZ,
+    deleted_at TIMESTAMPTZ,
+    written_at TIMESTAMPTZ,
+    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)
+);
+
+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);