diff --git a/internal/store/channel_memory_extraction_store.go b/internal/store/channel_memory_extraction_store.go new file mode 100644 index 00000000..d6f70c4b --- /dev/null +++ b/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 +} diff --git a/internal/store/pg/channel_memory_extraction.go b/internal/store/pg/channel_memory_extraction.go new file mode 100644 index 00000000..9a802855 --- /dev/null +++ b/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) +} diff --git a/internal/store/pg/factory.go b/internal/store/pg/factory.go index bde42a89..e00f346c 100644 --- a/internal/store/pg/factory.go +++ b/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), diff --git a/internal/store/sqlitestore/channel-memory-extraction.go b/internal/store/sqlitestore/channel-memory-extraction.go new file mode 100644 index 00000000..ff72def7 --- /dev/null +++ b/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) diff --git a/internal/store/sqlitestore/factory.go b/internal/store/sqlitestore/factory.go index 527a6447..78bcd55c 100644 --- a/internal/store/sqlitestore/factory.go +++ b/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), diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index 031f0c18..a818bed1 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 = 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, diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index 4423301b..950b868a 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/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 -- ============================================================ diff --git a/internal/store/stores.go b/internal/store/stores.go index 7a1a9887..a542a3da 100644 --- a/internal/store/stores.go +++ b/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 diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index 279d5bfd..7e689893 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 = 74 +const RequiredSchemaVersion uint = 75 diff --git a/migrations/000075_channel_memory_extraction.down.sql b/migrations/000075_channel_memory_extraction.down.sql new file mode 100644 index 00000000..2626649b --- /dev/null +++ b/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; diff --git a/migrations/000075_channel_memory_extraction.up.sql b/migrations/000075_channel_memory_extraction.up.sql new file mode 100644 index 00000000..e454b8a3 --- /dev/null +++ b/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);