diff --git a/pkg/memory/sqlc/archival.sql.go b/pkg/memory/sqlc/archival.sql.go index 6dccfc53d..b5ae8e33b 100644 --- a/pkg/memory/sqlc/archival.sql.go +++ b/pkg/memory/sqlc/archival.sql.go @@ -8,6 +8,7 @@ package sqlc import ( "context" "strings" + "time" "github.com/ZanzyTHEbar/dragonscale/pkg/ids" "github.com/ZanzyTHEbar/dragonscale/pkg/memory" @@ -76,6 +77,17 @@ type GetArchivalChunkParams struct { AgentID string `db:"agent_id" json:"agent_id"` } +type GetArchivalChunkRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + // GetArchivalChunk // // SELECT ac.id, @@ -91,9 +103,9 @@ type GetArchivalChunkParams struct { // WHERE ac.id = ?1 // AND ri.agent_id = ?2 // LIMIT 1 -func (q *Queries) GetArchivalChunk(ctx context.Context, arg GetArchivalChunkParams) (ArchivalChunk, error) { +func (q *Queries) GetArchivalChunk(ctx context.Context, arg GetArchivalChunkParams) (GetArchivalChunkRow, error) { row := q.db.QueryRowContext(ctx, GetArchivalChunk, arg.ID, arg.AgentID) - var i ArchivalChunk + var i GetArchivalChunkRow err := row.Scan( &i.ID, &i.RecallID, @@ -127,6 +139,17 @@ type GetArchivalChunksByIDsParams struct { AgentID string `db:"agent_id" json:"agent_id"` } +type GetArchivalChunksByIDsRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + // GetArchivalChunksByIDs // // SELECT ac.id, @@ -141,7 +164,7 @@ type GetArchivalChunksByIDsParams struct { // JOIN recall_items ri ON ac.recall_id = ri.id // WHERE ac.id IN (/*SLICE:ids*/?) // AND ri.agent_id = ?2 -func (q *Queries) GetArchivalChunksByIDs(ctx context.Context, arg GetArchivalChunksByIDsParams) ([]ArchivalChunk, error) { +func (q *Queries) GetArchivalChunksByIDs(ctx context.Context, arg GetArchivalChunksByIDsParams) ([]GetArchivalChunksByIDsRow, error) { query := GetArchivalChunksByIDs var queryParams []interface{} if len(arg.Ids) > 0 { @@ -158,9 +181,9 @@ func (q *Queries) GetArchivalChunksByIDs(ctx context.Context, arg GetArchivalChu return nil, err } defer rows.Close() - items := []ArchivalChunk{} + items := []GetArchivalChunksByIDsRow{} for rows.Next() { - var i ArchivalChunk + var i GetArchivalChunksByIDsRow if err := rows.Scan( &i.ID, &i.RecallID, @@ -218,6 +241,17 @@ type InsertArchivalChunkParams struct { Hash string `db:"hash" json:"hash"` } +type InsertArchivalChunkRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + // Archival Chunk queries // // INSERT INTO archival_chunks ( @@ -241,7 +275,7 @@ type InsertArchivalChunkParams struct { // datetime('now') // ) // RETURNING id, recall_id, chunk_index, content, embedding, source, hash, created_at -func (q *Queries) InsertArchivalChunk(ctx context.Context, arg InsertArchivalChunkParams) (ArchivalChunk, error) { +func (q *Queries) InsertArchivalChunk(ctx context.Context, arg InsertArchivalChunkParams) (InsertArchivalChunkRow, error) { row := q.db.QueryRowContext(ctx, InsertArchivalChunk, arg.ID, arg.RecallID, @@ -251,7 +285,7 @@ func (q *Queries) InsertArchivalChunk(ctx context.Context, arg InsertArchivalChu arg.Source, arg.Hash, ) - var i ArchivalChunk + var i InsertArchivalChunkRow err := row.Scan( &i.ID, &i.RecallID, @@ -287,6 +321,17 @@ type ListAllArchivalChunksParams struct { Lim int64 `db:"lim" json:"lim"` } +type ListAllArchivalChunksRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + // ListAllArchivalChunks // // SELECT ac.id, @@ -302,15 +347,15 @@ type ListAllArchivalChunksParams struct { // WHERE ri.agent_id = ?1 // ORDER BY ac.created_at DESC // LIMIT ?3 OFFSET ?2 -func (q *Queries) ListAllArchivalChunks(ctx context.Context, arg ListAllArchivalChunksParams) ([]ArchivalChunk, error) { +func (q *Queries) ListAllArchivalChunks(ctx context.Context, arg ListAllArchivalChunksParams) ([]ListAllArchivalChunksRow, error) { rows, err := q.db.QueryContext(ctx, ListAllArchivalChunks, arg.AgentID, arg.Off, arg.Lim) if err != nil { return nil, err } defer rows.Close() - items := []ArchivalChunk{} + items := []ListAllArchivalChunksRow{} for rows.Next() { - var i ArchivalChunk + var i ListAllArchivalChunksRow if err := rows.Scan( &i.ID, &i.RecallID, @@ -357,6 +402,17 @@ type ListArchivalChunksParams struct { Lim int64 `db:"lim" json:"lim"` } +type ListArchivalChunksRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + // ListArchivalChunks // // SELECT ac.id, @@ -373,15 +429,15 @@ type ListArchivalChunksParams struct { // AND ri.agent_id = ?2 // ORDER BY ac.chunk_index // LIMIT ?3 -func (q *Queries) ListArchivalChunks(ctx context.Context, arg ListArchivalChunksParams) ([]ArchivalChunk, error) { +func (q *Queries) ListArchivalChunks(ctx context.Context, arg ListArchivalChunksParams) ([]ListArchivalChunksRow, error) { rows, err := q.db.QueryContext(ctx, ListArchivalChunks, arg.RecallID, arg.AgentID, arg.Lim) if err != nil { return nil, err } defer rows.Close() - items := []ArchivalChunk{} + items := []ListArchivalChunksRow{} for rows.Next() { - var i ArchivalChunk + var i ListArchivalChunksRow if err := rows.Scan( &i.ID, &i.RecallID, diff --git a/pkg/memory/sqlc/consolidation.sql.go b/pkg/memory/sqlc/consolidation.sql.go new file mode 100644 index 000000000..ef40310ba --- /dev/null +++ b/pkg/memory/sqlc/consolidation.sql.go @@ -0,0 +1,210 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: consolidation.sql + +package sqlc + +import ( + "context" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" +) + +const FindSimilarRecallItems = `-- name: FindSimilarRecallItems :many +SELECT ri.id, + ri.agent_id, + ri.content, + ac.embedding as embedding, + vector_distance_cosine(ac.embedding, ?1) as distance +FROM recall_items ri + JOIN archival_chunks ac ON ac.recall_id = ri.id + AND ac.chunk_index = 0 +WHERE ri.agent_id = ?2 + AND ri.id != ?3 + AND ac.embedding IS NOT NULL +ORDER BY distance ASC +LIMIT ?4 +` + +type FindSimilarRecallItemsParams struct { + QueryEmbedding interface{} `db:"query_embedding" json:"query_embedding"` + AgentID string `db:"agent_id" json:"agent_id"` + ExcludeID ids.UUID `db:"exclude_id" json:"exclude_id"` + Lim int64 `db:"lim" json:"lim"` +} + +type FindSimilarRecallItemsRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Distance interface{} `db:"distance" json:"distance"` +} + +// Finds recall items with high cosine similarity to the given embedding. +// Uses libSQL's built-in vector similarity if available, otherwise returns +// candidates for manual comparison. +// +// SELECT ri.id, +// ri.agent_id, +// ri.content, +// ac.embedding as embedding, +// vector_distance_cosine(ac.embedding, ?1) as distance +// FROM recall_items ri +// JOIN archival_chunks ac ON ac.recall_id = ri.id +// AND ac.chunk_index = 0 +// WHERE ri.agent_id = ?2 +// AND ri.id != ?3 +// AND ac.embedding IS NOT NULL +// ORDER BY distance ASC +// LIMIT ?4 +func (q *Queries) FindSimilarRecallItems(ctx context.Context, arg FindSimilarRecallItemsParams) ([]FindSimilarRecallItemsRow, error) { + rows, err := q.db.QueryContext(ctx, FindSimilarRecallItems, + arg.QueryEmbedding, + arg.AgentID, + arg.ExcludeID, + arg.Lim, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []FindSimilarRecallItemsRow{} + for rows.Next() { + var i FindSimilarRecallItemsRow + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.Content, + &i.Embedding, + &i.Distance, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListRecallItemsForConsolidation = `-- name: ListRecallItemsForConsolidation :many +SELECT ri.id, + ri.agent_id, + ri.session_key, + ri.role, + ri.sector, + ri.importance, + ri.salience, + ri.decay_rate, + ri.content, + ri.tags, + ri.created_at, + ri.updated_at, + ac.embedding as embedding +FROM recall_items ri + LEFT JOIN archival_chunks ac ON ac.recall_id = ri.id + AND ac.chunk_index = 0 +WHERE ri.created_at > ?1 + AND ri.suppressed_at IS NULL + AND ( + ?2 = '' + OR ri.agent_id = ?2 + ) +ORDER BY ri.created_at DESC +LIMIT ?3 +` + +type ListRecallItemsForConsolidationParams struct { + Cutoff time.Time `db:"cutoff" json:"cutoff"` + AgentID interface{} `db:"agent_id" json:"agent_id"` + Lim int64 `db:"lim" json:"lim"` +} + +type ListRecallItemsForConsolidationRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` +} + +// Consolidation queries for memory graph maintenance (ADR-004) +// Returns recall items created after the cutoff time, joined with their first +// archival chunk's embedding for similarity comparison. +// +// SELECT ri.id, +// ri.agent_id, +// ri.session_key, +// ri.role, +// ri.sector, +// ri.importance, +// ri.salience, +// ri.decay_rate, +// ri.content, +// ri.tags, +// ri.created_at, +// ri.updated_at, +// ac.embedding as embedding +// FROM recall_items ri +// LEFT JOIN archival_chunks ac ON ac.recall_id = ri.id +// AND ac.chunk_index = 0 +// WHERE ri.created_at > ?1 +// AND ri.suppressed_at IS NULL +// AND ( +// ?2 = '' +// OR ri.agent_id = ?2 +// ) +// ORDER BY ri.created_at DESC +// LIMIT ?3 +func (q *Queries) ListRecallItemsForConsolidation(ctx context.Context, arg ListRecallItemsForConsolidationParams) ([]ListRecallItemsForConsolidationRow, error) { + rows, err := q.db.QueryContext(ctx, ListRecallItemsForConsolidation, arg.Cutoff, arg.AgentID, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListRecallItemsForConsolidationRow{} + for rows.Next() { + var i ListRecallItemsForConsolidationRow + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.SessionKey, + &i.Role, + &i.Sector, + &i.Importance, + &i.Salience, + &i.DecayRate, + &i.Content, + &i.Tags, + &i.CreatedAt, + &i.UpdatedAt, + &i.Embedding, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/pkg/memory/sqlc/immutable_messages.sql.go b/pkg/memory/sqlc/immutable_messages.sql.go new file mode 100644 index 000000000..38b85bda6 --- /dev/null +++ b/pkg/memory/sqlc/immutable_messages.sql.go @@ -0,0 +1,239 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: immutable_messages.sql + +package sqlc + +import ( + "context" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" +) + +const CountImmutableMessages = `-- name: CountImmutableMessages :one +SELECT COUNT(*) +FROM immutable_messages +WHERE session_key = ?1 +` + +type CountImmutableMessagesParams struct { + SessionKey string `db:"session_key" json:"session_key"` +} + +// CountImmutableMessages +// +// SELECT COUNT(*) +// FROM immutable_messages +// WHERE session_key = ?1 +func (q *Queries) CountImmutableMessages(ctx context.Context, arg CountImmutableMessagesParams) (int64, error) { + row := q.db.QueryRowContext(ctx, CountImmutableMessages, arg.SessionKey) + var count int64 + err := row.Scan(&count) + return count, err +} + +const GetImmutableMessage = `-- name: GetImmutableMessage :one +SELECT id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at +FROM immutable_messages +WHERE id = ?1 +LIMIT 1 +` + +type GetImmutableMessageParams struct { + ID ids.UUID `db:"id" json:"id"` +} + +// GetImmutableMessage +// +// SELECT id, +// session_key, +// role, +// content, +// tool_call_id, +// tool_calls, +// token_estimate, +// created_at +// FROM immutable_messages +// WHERE id = ?1 +// LIMIT 1 +func (q *Queries) GetImmutableMessage(ctx context.Context, arg GetImmutableMessageParams) (ImmutableMessage, error) { + row := q.db.QueryRowContext(ctx, GetImmutableMessage, arg.ID) + var i ImmutableMessage + err := row.Scan( + &i.ID, + &i.SessionKey, + &i.Role, + &i.Content, + &i.ToolCallID, + &i.ToolCalls, + &i.TokenEstimate, + &i.CreatedAt, + ) + return i, err +} + +const InsertImmutableMessage = `-- name: InsertImmutableMessage :one +INSERT INTO immutable_messages ( + id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate + ) +VALUES ( + ?1, + ?2, + ?3, + ?4, + ?5, + ?6, + ?7 + ) +RETURNING id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at +` + +type InsertImmutableMessageParams struct { + ID ids.UUID `db:"id" json:"id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Content string `db:"content" json:"content"` + ToolCallID string `db:"tool_call_id" json:"tool_call_id"` + ToolCalls string `db:"tool_calls" json:"tool_calls"` + TokenEstimate int64 `db:"token_estimate" json:"token_estimate"` +} + +// Immutable Messages queries (LCM ADR-001: append-only verbatim message store) +// +// INSERT INTO immutable_messages ( +// id, +// session_key, +// role, +// content, +// tool_call_id, +// tool_calls, +// token_estimate +// ) +// VALUES ( +// ?1, +// ?2, +// ?3, +// ?4, +// ?5, +// ?6, +// ?7 +// ) +// RETURNING id, +// session_key, +// role, +// content, +// tool_call_id, +// tool_calls, +// token_estimate, +// created_at +func (q *Queries) InsertImmutableMessage(ctx context.Context, arg InsertImmutableMessageParams) (ImmutableMessage, error) { + row := q.db.QueryRowContext(ctx, InsertImmutableMessage, + arg.ID, + arg.SessionKey, + arg.Role, + arg.Content, + arg.ToolCallID, + arg.ToolCalls, + arg.TokenEstimate, + ) + var i ImmutableMessage + err := row.Scan( + &i.ID, + &i.SessionKey, + &i.Role, + &i.Content, + &i.ToolCallID, + &i.ToolCalls, + &i.TokenEstimate, + &i.CreatedAt, + ) + return i, err +} + +const ListImmutableMessages = `-- name: ListImmutableMessages :many +SELECT id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at +FROM immutable_messages +WHERE session_key = ?1 +ORDER BY created_at ASC +LIMIT ?3 OFFSET ?2 +` + +type ListImmutableMessagesParams struct { + SessionKey string `db:"session_key" json:"session_key"` + Off int64 `db:"off" json:"off"` + Lim int64 `db:"lim" json:"lim"` +} + +// ListImmutableMessages +// +// SELECT id, +// session_key, +// role, +// content, +// tool_call_id, +// tool_calls, +// token_estimate, +// created_at +// FROM immutable_messages +// WHERE session_key = ?1 +// ORDER BY created_at ASC +// LIMIT ?3 OFFSET ?2 +func (q *Queries) ListImmutableMessages(ctx context.Context, arg ListImmutableMessagesParams) ([]ImmutableMessage, error) { + rows, err := q.db.QueryContext(ctx, ListImmutableMessages, arg.SessionKey, arg.Off, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ImmutableMessage{} + for rows.Next() { + var i ImmutableMessage + if err := rows.Scan( + &i.ID, + &i.SessionKey, + &i.Role, + &i.Content, + &i.ToolCallID, + &i.ToolCalls, + &i.TokenEstimate, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/pkg/memory/sqlc/memory_edges.sql.go b/pkg/memory/sqlc/memory_edges.sql.go new file mode 100644 index 000000000..59c049ba3 --- /dev/null +++ b/pkg/memory/sqlc/memory_edges.sql.go @@ -0,0 +1,235 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: memory_edges.sql + +package sqlc + +import ( + "context" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" +) + +const CountMemoryEdgesForItem = `-- name: CountMemoryEdgesForItem :one +SELECT COUNT(*) +FROM memory_edges +WHERE from_id = ?1 + OR to_id = ?1 +` + +type CountMemoryEdgesForItemParams struct { + MemoryID ids.UUID `db:"memory_id" json:"memory_id"` +} + +// CountMemoryEdgesForItem +// +// SELECT COUNT(*) +// FROM memory_edges +// WHERE from_id = ?1 +// OR to_id = ?1 +func (q *Queries) CountMemoryEdgesForItem(ctx context.Context, arg CountMemoryEdgesForItemParams) (int64, error) { + row := q.db.QueryRowContext(ctx, CountMemoryEdgesForItem, arg.MemoryID) + var count int64 + err := row.Scan(&count) + return count, err +} + +const DeleteMemoryEdgesForItem = `-- name: DeleteMemoryEdgesForItem :exec +DELETE FROM memory_edges +WHERE from_id = ?1 + OR to_id = ?1 +` + +type DeleteMemoryEdgesForItemParams struct { + MemoryID ids.UUID `db:"memory_id" json:"memory_id"` +} + +// DeleteMemoryEdgesForItem +// +// DELETE FROM memory_edges +// WHERE from_id = ?1 +// OR to_id = ?1 +func (q *Queries) DeleteMemoryEdgesForItem(ctx context.Context, arg DeleteMemoryEdgesForItemParams) error { + _, err := q.db.ExecContext(ctx, DeleteMemoryEdgesForItem, arg.MemoryID) + return err +} + +const InsertMemoryEdge = `-- name: InsertMemoryEdge :one +INSERT INTO memory_edges (from_id, to_id, edge_type, weight) +VALUES ( + ?1, + ?2, + ?3, + ?4 + ) +RETURNING id, + from_id, + to_id, + edge_type, + weight, + created_at +` + +type InsertMemoryEdgeParams struct { + FromID ids.UUID `db:"from_id" json:"from_id"` + ToID ids.UUID `db:"to_id" json:"to_id"` + EdgeType string `db:"edge_type" json:"edge_type"` + Weight float64 `db:"weight" json:"weight"` +} + +// Memory Graph Edges queries (ADR-004: typed relational memory) +// +// INSERT INTO memory_edges (from_id, to_id, edge_type, weight) +// VALUES ( +// ?1, +// ?2, +// ?3, +// ?4 +// ) +// RETURNING id, +// from_id, +// to_id, +// edge_type, +// weight, +// created_at +func (q *Queries) InsertMemoryEdge(ctx context.Context, arg InsertMemoryEdgeParams) (MemoryEdge, error) { + row := q.db.QueryRowContext(ctx, InsertMemoryEdge, + arg.FromID, + arg.ToID, + arg.EdgeType, + arg.Weight, + ) + var i MemoryEdge + err := row.Scan( + &i.ID, + &i.FromID, + &i.ToID, + &i.EdgeType, + &i.Weight, + &i.CreatedAt, + ) + return i, err +} + +const ListMemoryEdgesByType = `-- name: ListMemoryEdgesByType :many +SELECT id, + from_id, + to_id, + edge_type, + weight, + created_at +FROM memory_edges +WHERE edge_type = ?1 +ORDER BY created_at DESC +LIMIT ?2 +` + +type ListMemoryEdgesByTypeParams struct { + EdgeType string `db:"edge_type" json:"edge_type"` + Lim int64 `db:"lim" json:"lim"` +} + +// ListMemoryEdgesByType +// +// SELECT id, +// from_id, +// to_id, +// edge_type, +// weight, +// created_at +// FROM memory_edges +// WHERE edge_type = ?1 +// ORDER BY created_at DESC +// LIMIT ?2 +func (q *Queries) ListMemoryEdgesByType(ctx context.Context, arg ListMemoryEdgesByTypeParams) ([]MemoryEdge, error) { + rows, err := q.db.QueryContext(ctx, ListMemoryEdgesByType, arg.EdgeType, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MemoryEdge{} + for rows.Next() { + var i MemoryEdge + if err := rows.Scan( + &i.ID, + &i.FromID, + &i.ToID, + &i.EdgeType, + &i.Weight, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListMemoryEdgesForItem = `-- name: ListMemoryEdgesForItem :many +SELECT id, + from_id, + to_id, + edge_type, + weight, + created_at +FROM memory_edges +WHERE from_id = ?1 + OR to_id = ?1 +ORDER BY created_at DESC +LIMIT ?2 +` + +type ListMemoryEdgesForItemParams struct { + MemoryID ids.UUID `db:"memory_id" json:"memory_id"` + Lim int64 `db:"lim" json:"lim"` +} + +// ListMemoryEdgesForItem +// +// SELECT id, +// from_id, +// to_id, +// edge_type, +// weight, +// created_at +// FROM memory_edges +// WHERE from_id = ?1 +// OR to_id = ?1 +// ORDER BY created_at DESC +// LIMIT ?2 +func (q *Queries) ListMemoryEdgesForItem(ctx context.Context, arg ListMemoryEdgesForItemParams) ([]MemoryEdge, error) { + rows, err := q.db.QueryContext(ctx, ListMemoryEdgesForItem, arg.MemoryID, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MemoryEdge{} + for rows.Next() { + var i MemoryEdge + if err := rows.Scan( + &i.ID, + &i.FromID, + &i.ToID, + &i.EdgeType, + &i.Weight, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/pkg/memory/sqlc/models.go b/pkg/memory/sqlc/models.go index 2cae18c4f..64ea15044 100644 --- a/pkg/memory/sqlc/models.go +++ b/pkg/memory/sqlc/models.go @@ -181,14 +181,15 @@ type AgentToolResult struct { } type ArchivalChunk struct { - ID ids.UUID `db:"id" json:"id"` - RecallID ids.UUID `db:"recall_id" json:"recall_id"` - ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` - Content string `db:"content" json:"content"` - Embedding memory.Embedding `db:"embedding" json:"embedding"` - Source string `db:"source" json:"source"` - Hash string `db:"hash" json:"hash"` - CreatedAt time.Time `db:"created_at" json:"created_at"` + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + SuppressedAt *time.Time `db:"suppressed_at" json:"suppressed_at"` } type DagEdge struct { @@ -232,6 +233,17 @@ type DagSnapshot struct { UpdatedAt time.Time `db:"updated_at" json:"updated_at"` } +type ImmutableMessage struct { + ID ids.UUID `db:"id" json:"id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Content string `db:"content" json:"content"` + ToolCallID string `db:"tool_call_id" json:"tool_call_id"` + ToolCalls string `db:"tool_calls" json:"tool_calls"` + TokenEstimate int64 `db:"token_estimate" json:"token_estimate"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + type Job struct { ID ids.UUID `db:"id" json:"id"` Kind string `db:"kind" json:"kind"` @@ -284,6 +296,15 @@ type MapRun struct { CompletedAt *time.Time `db:"completed_at" json:"completed_at"` } +type MemoryEdge struct { + ID int64 `db:"id" json:"id"` + FromID ids.UUID `db:"from_id" json:"from_id"` + ToID ids.UUID `db:"to_id" json:"to_id"` + EdgeType string `db:"edge_type" json:"edge_type"` + Weight float64 `db:"weight" json:"weight"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + type MemorySummary struct { ID ids.UUID `db:"id" json:"id"` AgentID string `db:"agent_id" json:"agent_id"` @@ -294,19 +315,30 @@ type MemorySummary struct { CreatedAt time.Time `db:"created_at" json:"created_at"` } -type RecallItem struct { - ID ids.UUID `db:"id" json:"id"` - AgentID string `db:"agent_id" json:"agent_id"` - SessionKey string `db:"session_key" json:"session_key"` - Role string `db:"role" json:"role"` - Sector memory.Sector `db:"sector" json:"sector"` - Importance float64 `db:"importance" json:"importance"` - Salience float64 `db:"salience" json:"salience"` - DecayRate float64 `db:"decay_rate" json:"decay_rate"` - Content string `db:"content" json:"content"` - Tags string `db:"tags" json:"tags"` - CreatedAt time.Time `db:"created_at" json:"created_at"` - UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +type TaskBaseline struct { + AgentID string `db:"agent_id" json:"agent_id"` + Count *int64 `db:"count" json:"count"` + MeanTokens *int64 `db:"mean_tokens" json:"mean_tokens"` + MeanErrors *float64 `db:"mean_errors" json:"mean_errors"` + MeanUserCorrections *float64 `db:"mean_user_corrections" json:"mean_user_corrections"` + M2Tokens *float64 `db:"m2_tokens" json:"m2_tokens"` + M2Errors *float64 `db:"m2_errors" json:"m2_errors"` + M2UserCorrections *float64 `db:"m2_user_corrections" json:"m2_user_corrections"` + UpdatedAt *time.Time `db:"updated_at" json:"updated_at"` +} + +type TaskCompletion struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + ConversationID ids.UUID `db:"conversation_id" json:"conversation_id"` + RunID ids.UUID `db:"run_id" json:"run_id"` + Description string `db:"description" json:"description"` + TokensUsed *int64 `db:"tokens_used" json:"tokens_used"` + ToolCalls *int64 `db:"tool_calls" json:"tool_calls"` + Errors *int64 `db:"errors" json:"errors"` + UserCorrections *int64 `db:"user_corrections" json:"user_corrections"` + Completed bool `db:"completed" json:"completed"` + CreatedAt time.Time `db:"created_at" json:"created_at"` } type WorkingContext struct { diff --git a/pkg/memory/sqlc/querier.go b/pkg/memory/sqlc/querier.go index 9e886c4e2..15bc34d90 100644 --- a/pkg/memory/sqlc/querier.go +++ b/pkg/memory/sqlc/querier.go @@ -115,6 +115,13 @@ type Querier interface { // JOIN recall_items ri ON ac.recall_id = ri.id // WHERE ri.agent_id = ?1 CountArchivalChunks(ctx context.Context, arg CountArchivalChunksParams) (int64, error) + //CountArchivalChunksWithoutEmbedding + // + // SELECT COUNT(*) + // FROM archival_chunks + // WHERE embedding IS NULL + // AND suppressed_at IS NULL + CountArchivalChunksWithoutEmbedding(ctx context.Context) (int64, error) //CountAuditEntries // // SELECT COUNT(*) @@ -128,6 +135,12 @@ type Querier interface { // WHERE agent_id = ?1 // AND action = ?2 CountAuditEntriesByAction(ctx context.Context, arg CountAuditEntriesByActionParams) (int64, error) + //CountImmutableMessages + // + // SELECT COUNT(*) + // FROM immutable_messages + // WHERE session_key = ?1 + CountImmutableMessages(ctx context.Context, arg CountImmutableMessagesParams) (int64, error) //CountJobsByStatus // // SELECT status, @@ -149,11 +162,19 @@ type Querier interface { // WHERE run_id = ?1 // GROUP BY status CountMapItemsByRunAndStatus(ctx context.Context, arg CountMapItemsByRunAndStatusParams) ([]CountMapItemsByRunAndStatusRow, error) + //CountMemoryEdgesForItem + // + // SELECT COUNT(*) + // FROM memory_edges + // WHERE from_id = ?1 + // OR to_id = ?1 + CountMemoryEdgesForItem(ctx context.Context, arg CountMemoryEdgesForItemParams) (int64, error) //CountRecallItems // // SELECT COUNT(*) // FROM recall_items // WHERE agent_id = ?1 + // AND suppressed_at IS NULL // AND ( // session_key = ?2 // OR ?2 = '' @@ -221,6 +242,23 @@ type Querier interface { // VALUES (?, ?, ?, ?) // RETURNING id, conversation_id, title, metadata_json, created_at, updated_at CreateAgentThread(ctx context.Context, arg CreateAgentThreadParams) (AgentThread, error) + // Batch decay of recall item importance (Cortex decay task) + // + // UPDATE recall_items + // SET importance = MAX( + // recall_items.importance * ?1, + // ?2 + // ), + // updated_at = datetime('now') + // WHERE recall_items.id IN ( + // SELECT ri.id + // FROM recall_items ri + // WHERE ri.importance > ?2 + // AND ri.suppressed_at IS NULL + // ORDER BY ri.updated_at ASC + // LIMIT ?3 + // ) + DecayRecallImportanceBatch(ctx context.Context, arg DecayRecallImportanceBatchParams) error //DeleteAgentConversationLink // // DELETE FROM agent_conversation_links @@ -245,6 +283,12 @@ type Querier interface { // WHERE agent_id = ?1 // AND key = ?2 DeleteKV(ctx context.Context, arg DeleteKVParams) error + //DeleteMemoryEdgesForItem + // + // DELETE FROM memory_edges + // WHERE from_id = ?1 + // OR to_id = ?1 + DeleteMemoryEdgesForItem(ctx context.Context, arg DeleteMemoryEdgesForItemParams) error //DeleteRecallItem // // DELETE FROM recall_items @@ -295,6 +339,24 @@ type Querier interface { // created_at ASC // LIMIT 1 FindNextRunnableJob(ctx context.Context) (ids.UUID, error) + // Finds recall items with high cosine similarity to the given embedding. + // Uses libSQL's built-in vector similarity if available, otherwise returns + // candidates for manual comparison. + // + // SELECT ri.id, + // ri.agent_id, + // ri.content, + // ac.embedding as embedding, + // vector_distance_cosine(ac.embedding, ?1) as distance + // FROM recall_items ri + // JOIN archival_chunks ac ON ac.recall_id = ri.id + // AND ac.chunk_index = 0 + // WHERE ri.agent_id = ?2 + // AND ri.id != ?3 + // AND ac.embedding IS NOT NULL + // ORDER BY distance ASC + // LIMIT ?4 + FindSimilarRecallItems(ctx context.Context, arg FindSimilarRecallItemsParams) ([]FindSimilarRecallItemsRow, error) //GetAgentCheckpointByConversationIDAndName // // SELECT id, conversation_id, name, run_state_id, metadata_json, created_at, updated_at @@ -354,7 +416,7 @@ type Querier interface { // WHERE ac.id = ?1 // AND ri.agent_id = ?2 // LIMIT 1 - GetArchivalChunk(ctx context.Context, arg GetArchivalChunkParams) (ArchivalChunk, error) + GetArchivalChunk(ctx context.Context, arg GetArchivalChunkParams) (GetArchivalChunkRow, error) //GetArchivalChunksByIDs // // SELECT ac.id, @@ -369,7 +431,26 @@ type Querier interface { // JOIN recall_items ri ON ac.recall_id = ri.id // WHERE ac.id IN (/*SLICE:ids*/?) // AND ri.agent_id = ?2 - GetArchivalChunksByIDs(ctx context.Context, arg GetArchivalChunksByIDsParams) ([]ArchivalChunk, error) + GetArchivalChunksByIDs(ctx context.Context, arg GetArchivalChunksByIDsParams) ([]GetArchivalChunksByIDsRow, error) + // Get tasks completed since the given time for RL processing + // + // SELECT id, + // agent_id, + // conversation_id, + // run_id, + // description, + // tokens_used, + // tool_calls, + // errors, + // user_corrections, + // completed, + // created_at + // FROM task_completions + // WHERE agent_id = ?1 + // AND created_at > ?2 + // AND completed = 1 + // ORDER BY created_at ASC + GetCompletedTasks(ctx context.Context, arg GetCompletedTasksParams) ([]TaskCompletion, error) //GetDAGNodeBySnapshotAndNodeID // // SELECT id, @@ -423,6 +504,20 @@ type Querier interface { // AND name = ?2 // LIMIT 1 GetDocument(ctx context.Context, arg GetDocumentParams) (AgentDocument, error) + //GetImmutableMessage + // + // SELECT id, + // session_key, + // role, + // content, + // tool_call_id, + // tool_calls, + // token_estimate, + // created_at + // FROM immutable_messages + // WHERE id = ?1 + // LIMIT 1 + GetImmutableMessage(ctx context.Context, arg GetImmutableMessageParams) (ImmutableMessage, error) //GetJob // // SELECT id, kind, status, run_at, attempts, max_attempts, locked_at, locked_by, payload_json, dedupe_key, last_error, created_at, updated_at, completed_at @@ -507,6 +602,30 @@ type Querier interface { // AND idempotency_key = ?4 // LIMIT 1 GetMapRunByIdempotencyKey(ctx context.Context, arg GetMapRunByIdempotencyKeyParams) (MapRun, error) + // Get memories ordered by their task retrieval count (for RL analysis) + // + // SELECT id, + // agent_id, + // session_key, + // role, + // sector, + // importance, + // salience, + // decay_rate, + // content, + // tags, + // rl_weight, + // rl_credit, + // self_report_score, + // task_retrieval_count, + // created_at, + // updated_at + // FROM recall_items + // WHERE agent_id = ?1 + // AND suppressed_at IS NULL + // ORDER BY task_retrieval_count DESC + // LIMIT ?2 + GetMemoriesByRetrievalCount(ctx context.Context, arg GetMemoriesByRetrievalCountParams) ([]GetMemoriesByRetrievalCountRow, error) //GetRecallItem // // SELECT id, @@ -524,8 +643,9 @@ type Querier interface { // FROM recall_items // WHERE id = ?1 // AND agent_id = ?2 + // AND suppressed_at IS NULL // LIMIT 1 - GetRecallItem(ctx context.Context, arg GetRecallItemParams) (RecallItem, error) + GetRecallItem(ctx context.Context, arg GetRecallItemParams) (GetRecallItemRow, error) //GetRecallItemsByIDs // // SELECT id, @@ -543,7 +663,35 @@ type Querier interface { // FROM recall_items // WHERE id IN (/*SLICE:ids*/?) // AND agent_id = ?2 - GetRecallItemsByIDs(ctx context.Context, arg GetRecallItemsByIDsParams) ([]RecallItem, error) + // AND suppressed_at IS NULL + GetRecallItemsByIDs(ctx context.Context, arg GetRecallItemsByIDsParams) ([]GetRecallItemsByIDsRow, error) + // Get memories retrieved during a specific task with their self-report scores + // + // SELECT tr.memory_id, + // tr.similarity, + // ri.self_report_score + // FROM task_retrievals tr + // JOIN recall_items ri ON tr.memory_id = ri.id + // WHERE tr.task_id = ?1 + GetRetrievedMemories(ctx context.Context, arg GetRetrievedMemoriesParams) ([]GetRetrievedMemoriesRow, error) + // RL (Reinforcement Learning) queries for Memelord integration + // Task baseline queries for per-agent performance statistics + // Get the baseline statistics for an agent + // + // + // SELECT agent_id, + // count, + // mean_tokens, + // mean_errors, + // mean_user_corrections, + // m2_tokens, + // m2_errors, + // m2_user_corrections, + // updated_at + // FROM task_baselines + // WHERE agent_id = ?1 + // LIMIT 1 + GetTaskBaseline(ctx context.Context, arg GetTaskBaselineParams) (TaskBaseline, error) // Working Context queries // // SELECT agent_id, @@ -555,6 +703,30 @@ type Querier interface { // AND session_key = ?2 // LIMIT 1 GetWorkingContext(ctx context.Context, arg GetWorkingContextParams) (WorkingContext, error) + // Permanently deletes archival chunks for a recall item. + // + // DELETE FROM archival_chunks + // WHERE recall_id = ?1 + HardDeleteArchivalChunks(ctx context.Context, arg HardDeleteArchivalChunksParams) error + // Permanently deletes a single archival chunk by ID. + // + // DELETE FROM archival_chunks + // WHERE id = ?1 + HardDeleteChunk(ctx context.Context, arg HardDeleteChunkParams) error + // Permanently deletes a recall item (after quarantine period). + // + // DELETE FROM recall_items + // WHERE id = ?1 + // AND agent_id = ?2 + HardDeleteRecallItem(ctx context.Context, arg HardDeleteRecallItemParams) error + // Increment the task retrieval counter for a memory item + // + // UPDATE recall_items + // SET task_retrieval_count = task_retrieval_count + 1, + // updated_at = datetime('now') + // WHERE id = ?1 + // AND agent_id = ?2 + IncrementTaskRetrievalCount(ctx context.Context, arg IncrementTaskRetrievalCountParams) error // Archival Chunk queries // // INSERT INTO archival_chunks ( @@ -578,7 +750,7 @@ type Querier interface { // datetime('now') // ) // RETURNING id, recall_id, chunk_index, content, embedding, source, hash, created_at - InsertArchivalChunk(ctx context.Context, arg InsertArchivalChunkParams) (ArchivalChunk, error) + InsertArchivalChunk(ctx context.Context, arg InsertArchivalChunkParams) (InsertArchivalChunkRow, error) // Agent Audit Log queries // // INSERT INTO agent_audit_log ( @@ -702,6 +874,35 @@ type Querier interface { // created_at, // updated_at InsertDAGSnapshot(ctx context.Context, arg InsertDAGSnapshotParams) (DagSnapshot, error) + // Immutable Messages queries (LCM ADR-001: append-only verbatim message store) + // + // INSERT INTO immutable_messages ( + // id, + // session_key, + // role, + // content, + // tool_call_id, + // tool_calls, + // token_estimate + // ) + // VALUES ( + // ?1, + // ?2, + // ?3, + // ?4, + // ?5, + // ?6, + // ?7 + // ) + // RETURNING id, + // session_key, + // role, + // content, + // tool_call_id, + // tool_calls, + // token_estimate, + // created_at + InsertImmutableMessage(ctx context.Context, arg InsertImmutableMessageParams) (ImmutableMessage, error) //InsertMapItem // // INSERT INTO map_items ( @@ -764,6 +965,22 @@ type Querier interface { // ) // RETURNING id, agent_id, session_key, operator_kind, idempotency_key, status, total_items, queued_items, running_items, succeeded_items, failed_items, spec_fb, last_error, created_at, updated_at, completed_at InsertMapRun(ctx context.Context, arg InsertMapRunParams) (MapRun, error) + // Memory Graph Edges queries (ADR-004: typed relational memory) + // + // INSERT INTO memory_edges (from_id, to_id, edge_type, weight) + // VALUES ( + // ?1, + // ?2, + // ?3, + // ?4 + // ) + // RETURNING id, + // from_id, + // to_id, + // edge_type, + // weight, + // created_at + InsertMemoryEdge(ctx context.Context, arg InsertMemoryEdgeParams) (MemoryEdge, error) // Recall Item queries // // INSERT INTO recall_items ( @@ -777,6 +994,7 @@ type Querier interface { // decay_rate, // content, // tags, + // rl_weight, // created_at, // updated_at // ) @@ -791,6 +1009,7 @@ type Querier interface { // ?8, // ?9, // ?10, + // ?11, // datetime('now'), // datetime('now') // ) @@ -804,9 +1023,10 @@ type Querier interface { // decay_rate, // content, // tags, + // rl_weight, // created_at, // updated_at - InsertRecallItem(ctx context.Context, arg InsertRecallItemParams) (RecallItem, error) + InsertRecallItem(ctx context.Context, arg InsertRecallItemParams) (InsertRecallItemRow, error) //InsertSessionMessage // // INSERT INTO recall_items ( @@ -849,7 +1069,7 @@ type Querier interface { // tags, // created_at, // updated_at - InsertSessionMessage(ctx context.Context, arg InsertSessionMessageParams) (RecallItem, error) + InsertSessionMessage(ctx context.Context, arg InsertSessionMessageParams) (InsertSessionMessageRow, error) // Memory Summary queries // // INSERT INTO memory_summaries ( @@ -1012,7 +1232,7 @@ type Querier interface { // WHERE ri.agent_id = ?1 // ORDER BY ac.created_at DESC // LIMIT ?3 OFFSET ?2 - ListAllArchivalChunks(ctx context.Context, arg ListAllArchivalChunksParams) ([]ArchivalChunk, error) + ListAllArchivalChunks(ctx context.Context, arg ListAllArchivalChunksParams) ([]ListAllArchivalChunksRow, error) //ListAllDocuments // // SELECT id, @@ -1047,7 +1267,22 @@ type Querier interface { // AND ri.agent_id = ?2 // ORDER BY ac.chunk_index // LIMIT ?3 - ListArchivalChunks(ctx context.Context, arg ListArchivalChunksParams) ([]ArchivalChunk, error) + ListArchivalChunks(ctx context.Context, arg ListArchivalChunksParams) ([]ListArchivalChunksRow, error) + //ListArchivalChunksWithoutEmbedding + // + // SELECT id, + // recall_id, + // chunk_index, + // content, + // embedding, + // source, + // hash, + // created_at + // FROM archival_chunks + // WHERE embedding IS NULL + // AND suppressed_at IS NULL + // LIMIT ?1 + ListArchivalChunksWithoutEmbedding(ctx context.Context, arg ListArchivalChunksWithoutEmbeddingParams) ([]ListArchivalChunksWithoutEmbeddingRow, error) //ListAuditEntries // // SELECT id, @@ -1151,6 +1386,49 @@ type Querier interface { // ORDER BY name // LIMIT ?3 ListDocumentsByCategory(ctx context.Context, arg ListDocumentsByCategoryParams) ([]AgentDocument, error) + // List memories with high RL weights (credits) for priority retention + // + // SELECT id, + // agent_id, + // session_key, + // role, + // sector, + // importance, + // salience, + // decay_rate, + // content, + // tags, + // rl_weight, + // rl_credit, + // self_report_score, + // task_retrieval_count, + // created_at, + // updated_at + // FROM recall_items + // WHERE agent_id = ?1 + // AND suppressed_at IS NULL + // AND ( + // rl_credit > ?2 + // OR rl_weight > ?3 + // ) + // ORDER BY rl_credit DESC NULLS LAST + // LIMIT ?4 + ListHighValueMemories(ctx context.Context, arg ListHighValueMemoriesParams) ([]ListHighValueMemoriesRow, error) + //ListImmutableMessages + // + // SELECT id, + // session_key, + // role, + // content, + // tool_call_id, + // tool_calls, + // token_estimate, + // created_at + // FROM immutable_messages + // WHERE session_key = ?1 + // ORDER BY created_at ASC + // LIMIT ?3 OFFSET ?2 + ListImmutableMessages(ctx context.Context, arg ListImmutableMessagesParams) ([]ImmutableMessage, error) //ListJobs // // SELECT id, kind, status, run_at, attempts, max_attempts, locked_at, locked_by, payload_json, dedupe_key, last_error, created_at, updated_at, completed_at @@ -1187,6 +1465,73 @@ type Querier interface { // ORDER BY created_at DESC // LIMIT ?4 OFFSET ?3 ListMapRunsBySession(ctx context.Context, arg ListMapRunsBySessionParams) ([]MapRun, error) + //ListMemoryEdgesByType + // + // SELECT id, + // from_id, + // to_id, + // edge_type, + // weight, + // created_at + // FROM memory_edges + // WHERE edge_type = ?1 + // ORDER BY created_at DESC + // LIMIT ?2 + ListMemoryEdgesByType(ctx context.Context, arg ListMemoryEdgesByTypeParams) ([]MemoryEdge, error) + //ListMemoryEdgesForItem + // + // SELECT id, + // from_id, + // to_id, + // edge_type, + // weight, + // created_at + // FROM memory_edges + // WHERE from_id = ?1 + // OR to_id = ?1 + // ORDER BY created_at DESC + // LIMIT ?2 + ListMemoryEdgesForItem(ctx context.Context, arg ListMemoryEdgesForItemParams) ([]MemoryEdge, error) + // Returns archival chunks that have been soft-deleted and are + // eligible for permanent deletion. + // + // SELECT id, + // recall_id, + // chunk_index, + // content, + // embedding, + // source, + // hash, + // created_at, + // suppressed_at + // FROM archival_chunks + // WHERE suppressed_at IS NOT NULL + // AND suppressed_at < ?1 + // ORDER BY suppressed_at ASC + // LIMIT ?2 + ListQuarantinedArchivalChunks(ctx context.Context, arg ListQuarantinedArchivalChunksParams) ([]ArchivalChunk, error) + // Returns recall items that have been soft-deleted (suppressed) and are + // eligible for permanent deletion (older than quarantine period). + // + // SELECT id, + // agent_id, + // session_key, + // role, + // sector, + // importance, + // salience, + // decay_rate, + // content, + // tags, + // created_at, + // updated_at, + // suppressed_at + // FROM recall_items + // WHERE suppressed_at IS NOT NULL + // AND suppressed_at < ?1 + // ORDER BY suppressed_at ASC + // LIMIT ?2 + ListQuarantinedRecallItems(ctx context.Context, arg ListQuarantinedRecallItemsParams) ([]ListQuarantinedRecallItemsRow, error) //ListRecallItems // // SELECT id, @@ -1203,13 +1548,43 @@ type Querier interface { // updated_at // FROM recall_items // WHERE agent_id = ?1 + // AND suppressed_at IS NULL // AND ( // session_key = ?2 // OR ?2 = '' // ) // ORDER BY created_at DESC // LIMIT ?4 OFFSET ?3 - ListRecallItems(ctx context.Context, arg ListRecallItemsParams) ([]RecallItem, error) + ListRecallItems(ctx context.Context, arg ListRecallItemsParams) ([]ListRecallItemsRow, error) + // Consolidation queries for memory graph maintenance (ADR-004) + // Returns recall items created after the cutoff time, joined with their first + // archival chunk's embedding for similarity comparison. + // + // SELECT ri.id, + // ri.agent_id, + // ri.session_key, + // ri.role, + // ri.sector, + // ri.importance, + // ri.salience, + // ri.decay_rate, + // ri.content, + // ri.tags, + // ri.created_at, + // ri.updated_at, + // ac.embedding as embedding + // FROM recall_items ri + // LEFT JOIN archival_chunks ac ON ac.recall_id = ri.id + // AND ac.chunk_index = 0 + // WHERE ri.created_at > ?1 + // AND ri.suppressed_at IS NULL + // AND ( + // ?2 = '' + // OR ri.agent_id = ?2 + // ) + // ORDER BY ri.created_at DESC + // LIMIT ?3 + ListRecallItemsForConsolidation(ctx context.Context, arg ListRecallItemsForConsolidationParams) ([]ListRecallItemsForConsolidationRow, error) //ListSessionMessages // // SELECT id, @@ -1234,7 +1609,7 @@ type Querier interface { // ) // ORDER BY created_at ASC // LIMIT ?4 - ListSessionMessages(ctx context.Context, arg ListSessionMessagesParams) ([]RecallItem, error) + ListSessionMessages(ctx context.Context, arg ListSessionMessagesParams) ([]ListSessionMessagesRow, error) //ListSessionMessagesPaged // // SELECT id, @@ -1259,7 +1634,7 @@ type Querier interface { // ) // ORDER BY created_at ASC // LIMIT ?5 OFFSET ?4 - ListSessionMessagesPaged(ctx context.Context, arg ListSessionMessagesPagedParams) ([]RecallItem, error) + ListSessionMessagesPaged(ctx context.Context, arg ListSessionMessagesPagedParams) ([]ListSessionMessagesPagedRow, error) //ListSummaries // // SELECT id, @@ -1352,6 +1727,13 @@ type Querier interface { // WHERE id = ?3 // RETURNING id, kind, status, run_at, attempts, max_attempts, locked_at, locked_by, payload_json, dedupe_key, last_error, created_at, updated_at, completed_at RequeueJob(ctx context.Context, arg RequeueJobParams) (Job, error) + // Restores a soft-deleted recall item by clearing suppressed_at. + // + // UPDATE recall_items + // SET suppressed_at = NULL + // WHERE id = ?1 + // AND agent_id = ?2 + RestoreRecallItem(ctx context.Context, arg RestoreRecallItemParams) error //SearchRecallByKeyword // // SELECT ri.id, @@ -1369,9 +1751,69 @@ type Querier interface { // FROM recall_items ri // WHERE ri.content LIKE '%' || ?1 || '%' // AND ri.agent_id = ?2 + // AND ri.suppressed_at IS NULL // ORDER BY ri.importance DESC // LIMIT ?3 - SearchRecallByKeyword(ctx context.Context, arg SearchRecallByKeywordParams) ([]RecallItem, error) + SearchRecallByKeyword(ctx context.Context, arg SearchRecallByKeywordParams) ([]SearchRecallByKeywordRow, error) + // Sets suppressed_at for all chunks belonging to a recall item. + // + // UPDATE archival_chunks + // SET suppressed_at = datetime('now') + // WHERE recall_id = ?1 + SoftDeleteArchivalChunks(ctx context.Context, arg SoftDeleteArchivalChunksParams) error + // Soft delete queries (T2.4: quarantine before permanent deletion) + // Sets suppressed_at for a recall item instead of permanently deleting it. + // + // UPDATE recall_items + // SET suppressed_at = datetime('now') + // WHERE id = ?1 + // AND agent_id = ?2 + SoftDeleteRecallItem(ctx context.Context, arg SoftDeleteRecallItemParams) error + // Store a task completion record for RL analysis + // + // INSERT INTO task_completions ( + // id, + // agent_id, + // conversation_id, + // run_id, + // description, + // tokens_used, + // tool_calls, + // errors, + // user_corrections, + // completed + // ) + // VALUES ( + // ?1, + // ?2, + // ?3, + // ?4, + // ?5, + // ?6, + // ?7, + // ?8, + // ?9, + // ?10 + // ) + // RETURNING id, + // agent_id, + // conversation_id, + // run_id, + // description, + // tokens_used, + // tool_calls, + // errors, + // user_corrections, + // completed, + // created_at + StoreTaskCompletion(ctx context.Context, arg StoreTaskCompletionParams) (TaskCompletion, error) + // Store a memory retrieval record for a task + // + // INSERT INTO task_retrievals (id, task_id, memory_id, similarity) + // VALUES (?1, ?2, ?3, ?4) + // ON CONFLICT (task_id, memory_id) DO UPDATE SET + // similarity = excluded.similarity + StoreTaskRetrieval(ctx context.Context, arg StoreTaskRetrievalParams) error //UpdateAgentConversationTitle // // UPDATE agent_conversations @@ -1396,6 +1838,12 @@ type Querier interface { // WHERE id = ? // RETURNING id, conversation_id, status, metadata_json, created_at, updated_at UpdateAgentRunStatus(ctx context.Context, arg UpdateAgentRunStatusParams) (AgentRun, error) + //UpdateArchivalChunkEmbedding + // + // UPDATE archival_chunks + // SET embedding = ?1 + // WHERE id = ?2 + UpdateArchivalChunkEmbedding(ctx context.Context, arg UpdateArchivalChunkEmbeddingParams) error //UpdateMapRunProgress // // UPDATE map_runs @@ -1410,6 +1858,23 @@ type Querier interface { // WHERE id = ?8 // RETURNING id, agent_id, session_key, operator_kind, idempotency_key, status, total_items, queued_items, running_items, succeeded_items, failed_items, spec_fb, last_error, created_at, updated_at, completed_at UpdateMapRunProgress(ctx context.Context, arg UpdateMapRunProgressParams) (MapRun, error) + // Update the self-reported score for a memory item + // + // UPDATE recall_items + // SET self_report_score = ?1, + // updated_at = datetime('now') + // WHERE id = ?2 + // AND agent_id = ?3 + UpdateMemorySelfReportScore(ctx context.Context, arg UpdateMemorySelfReportScoreParams) error + // Update the RL weight and credit for a specific memory item + // + // UPDATE recall_items + // SET rl_weight = ?1, + // rl_credit = ?2, + // updated_at = datetime('now') + // WHERE id = ?3 + // AND agent_id = ?4 + UpdateMemoryWeight(ctx context.Context, arg UpdateMemoryWeightParams) error //UpdateRecallItem // // UPDATE recall_items @@ -1424,6 +1889,41 @@ type Querier interface { // WHERE id = ?8 // AND agent_id = ?9 UpdateRecallItem(ctx context.Context, arg UpdateRecallItemParams) error + // Insert or replace task baseline statistics for an agent + // + // INSERT INTO task_baselines ( + // agent_id, + // count, + // mean_tokens, + // mean_errors, + // mean_user_corrections, + // m2_tokens, + // m2_errors, + // m2_user_corrections, + // updated_at + // ) + // VALUES ( + // ?1, + // ?2, + // ?3, + // ?4, + // ?5, + // ?6, + // ?7, + // ?8, + // datetime('now') + // ) + // ON CONFLICT (agent_id) DO + // UPDATE + // SET count = excluded.count, + // mean_tokens = excluded.mean_tokens, + // mean_errors = excluded.mean_errors, + // mean_user_corrections = excluded.mean_user_corrections, + // m2_tokens = excluded.m2_tokens, + // m2_errors = excluded.m2_errors, + // m2_user_corrections = excluded.m2_user_corrections, + // updated_at = excluded.updated_at + UpdateTaskBaseline(ctx context.Context, arg UpdateTaskBaselineParams) error //UpsertDocument // // INSERT INTO agent_documents ( diff --git a/pkg/memory/sqlc/queries/consolidation.sql b/pkg/memory/sqlc/queries/consolidation.sql new file mode 100644 index 000000000..f476e6629 --- /dev/null +++ b/pkg/memory/sqlc/queries/consolidation.sql @@ -0,0 +1,45 @@ +-- Consolidation queries for memory graph maintenance (ADR-004) +-- name: ListRecallItemsForConsolidation :many +-- Returns recall items created after the cutoff time, joined with their first +-- archival chunk's embedding for similarity comparison. +SELECT ri.id, + ri.agent_id, + ri.session_key, + ri.role, + ri.sector, + ri.importance, + ri.salience, + ri.decay_rate, + ri.content, + ri.tags, + ri.created_at, + ri.updated_at, + ac.embedding as embedding +FROM recall_items ri + LEFT JOIN archival_chunks ac ON ac.recall_id = ri.id + AND ac.chunk_index = 0 +WHERE ri.created_at > sqlc.arg(cutoff) + AND ri.suppressed_at IS NULL + AND ( + sqlc.arg(agent_id) = '' + OR ri.agent_id = sqlc.arg(agent_id) + ) +ORDER BY ri.created_at DESC +LIMIT sqlc.arg(lim); +-- name: FindSimilarRecallItems :many +-- Finds recall items with high cosine similarity to the given embedding. +-- Uses libSQL's built-in vector similarity if available, otherwise returns +-- candidates for manual comparison. +SELECT ri.id, + ri.agent_id, + ri.content, + ac.embedding as embedding, + vector_distance_cosine(ac.embedding, sqlc.arg(query_embedding)) as distance +FROM recall_items ri + JOIN archival_chunks ac ON ac.recall_id = ri.id + AND ac.chunk_index = 0 +WHERE ri.agent_id = sqlc.arg(agent_id) + AND ri.id != sqlc.arg(exclude_id) + AND ac.embedding IS NOT NULL +ORDER BY distance ASC +LIMIT sqlc.arg(lim); \ No newline at end of file diff --git a/pkg/memory/sqlc/queries/immutable_messages.sql b/pkg/memory/sqlc/queries/immutable_messages.sql new file mode 100644 index 000000000..8fcd57845 --- /dev/null +++ b/pkg/memory/sqlc/queries/immutable_messages.sql @@ -0,0 +1,57 @@ +-- Immutable Messages queries (LCM ADR-001: append-only verbatim message store) +-- name: InsertImmutableMessage :one +INSERT INTO immutable_messages ( + id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate + ) +VALUES ( + sqlc.arg(id), + sqlc.arg(session_key), + sqlc.arg(role), + sqlc.arg(content), + sqlc.arg(tool_call_id), + sqlc.arg(tool_calls), + sqlc.arg(token_estimate) + ) +RETURNING id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at; +-- name: GetImmutableMessage :one +SELECT id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at +FROM immutable_messages +WHERE id = sqlc.arg(id) +LIMIT 1; +-- name: ListImmutableMessages :many +SELECT id, + session_key, + role, + content, + tool_call_id, + tool_calls, + token_estimate, + created_at +FROM immutable_messages +WHERE session_key = sqlc.arg(session_key) +ORDER BY created_at ASC +LIMIT sqlc.arg(lim) OFFSET sqlc.arg(off); +-- name: CountImmutableMessages :one +SELECT COUNT(*) +FROM immutable_messages +WHERE session_key = sqlc.arg(session_key); \ No newline at end of file diff --git a/pkg/memory/sqlc/queries/memory_edges.sql b/pkg/memory/sqlc/queries/memory_edges.sql new file mode 100644 index 000000000..13624dd50 --- /dev/null +++ b/pkg/memory/sqlc/queries/memory_edges.sql @@ -0,0 +1,47 @@ +-- Memory Graph Edges queries (ADR-004: typed relational memory) +-- name: InsertMemoryEdge :one +INSERT INTO memory_edges (from_id, to_id, edge_type, weight) +VALUES ( + sqlc.arg(from_id), + sqlc.arg(to_id), + sqlc.arg(edge_type), + sqlc.arg(weight) + ) +RETURNING id, + from_id, + to_id, + edge_type, + weight, + created_at; +-- name: ListMemoryEdgesForItem :many +SELECT id, + from_id, + to_id, + edge_type, + weight, + created_at +FROM memory_edges +WHERE from_id = sqlc.arg(memory_id) + OR to_id = sqlc.arg(memory_id) +ORDER BY created_at DESC +LIMIT sqlc.arg(lim); +-- name: CountMemoryEdgesForItem :one +SELECT COUNT(*) +FROM memory_edges +WHERE from_id = sqlc.arg(memory_id) + OR to_id = sqlc.arg(memory_id); +-- name: ListMemoryEdgesByType :many +SELECT id, + from_id, + to_id, + edge_type, + weight, + created_at +FROM memory_edges +WHERE edge_type = sqlc.arg(edge_type) +ORDER BY created_at DESC +LIMIT sqlc.arg(lim); +-- name: DeleteMemoryEdgesForItem :exec +DELETE FROM memory_edges +WHERE from_id = sqlc.arg(memory_id) + OR to_id = sqlc.arg(memory_id); \ No newline at end of file diff --git a/pkg/memory/sqlc/queries/recall.sql b/pkg/memory/sqlc/queries/recall.sql index 8ef82039e..c59fdf6be 100644 --- a/pkg/memory/sqlc/queries/recall.sql +++ b/pkg/memory/sqlc/queries/recall.sql @@ -11,6 +11,7 @@ INSERT INTO recall_items ( decay_rate, content, tags, + rl_weight, created_at, updated_at ) @@ -25,6 +26,7 @@ VALUES ( sqlc.arg(decay_rate), sqlc.arg(content), sqlc.arg(tags), + sqlc.arg(rl_weight), datetime('now'), datetime('now') ) @@ -38,6 +40,7 @@ RETURNING id, decay_rate, content, tags, + rl_weight, created_at, updated_at; -- name: GetRecallItem :one @@ -56,6 +59,7 @@ SELECT id, FROM recall_items WHERE id = sqlc.arg(id) AND agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL LIMIT 1; -- name: UpdateRecallItem :exec UPDATE recall_items @@ -88,6 +92,7 @@ SELECT id, updated_at FROM recall_items WHERE agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL AND ( session_key = sqlc.arg(session_key) OR sqlc.arg(session_key) = '' @@ -110,12 +115,14 @@ SELECT ri.id, FROM recall_items ri WHERE ri.content LIKE '%' || sqlc.arg(keyword) || '%' AND ri.agent_id = sqlc.arg(agent_id) + AND ri.suppressed_at IS NULL ORDER BY ri.importance DESC LIMIT sqlc.arg(lim); -- name: CountRecallItems :one SELECT COUNT(*) FROM recall_items WHERE agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL AND ( session_key = sqlc.arg(session_key) OR sqlc.arg(session_key) = '' @@ -135,7 +142,8 @@ SELECT id, updated_at FROM recall_items WHERE id IN (sqlc.slice('ids')) - AND agent_id = sqlc.arg(agent_id); + AND agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL; -- name: InsertSessionMessage :one INSERT INTO recall_items ( id, diff --git a/pkg/memory/sqlc/queries/recall_decay.sql b/pkg/memory/sqlc/queries/recall_decay.sql new file mode 100644 index 000000000..e2d4b5c53 --- /dev/null +++ b/pkg/memory/sqlc/queries/recall_decay.sql @@ -0,0 +1,38 @@ +-- Batch decay of recall item importance (Cortex decay task) +-- name: DecayRecallImportanceBatch :exec +UPDATE recall_items +SET importance = MAX( + recall_items.importance * sqlc.arg(factor), + sqlc.arg(floor_val) + ), + updated_at = datetime('now') +WHERE recall_items.id IN ( + SELECT ri.id + FROM recall_items ri + WHERE ri.importance > sqlc.arg(floor_val) + AND ri.suppressed_at IS NULL + ORDER BY ri.updated_at ASC + LIMIT sqlc.arg(batch_size) + ); +-- name: CountArchivalChunksWithoutEmbedding :one +SELECT COUNT(*) +FROM archival_chunks +WHERE embedding IS NULL + AND suppressed_at IS NULL; +-- name: ListArchivalChunksWithoutEmbedding :many +SELECT id, + recall_id, + chunk_index, + content, + embedding, + source, + hash, + created_at +FROM archival_chunks +WHERE embedding IS NULL + AND suppressed_at IS NULL +LIMIT sqlc.arg(lim); +-- name: UpdateArchivalChunkEmbedding :exec +UPDATE archival_chunks +SET embedding = sqlc.arg(embedding) +WHERE id = sqlc.arg(id); \ No newline at end of file diff --git a/pkg/memory/sqlc/queries/rl.sql b/pkg/memory/sqlc/queries/rl.sql new file mode 100644 index 000000000..03932e5cf --- /dev/null +++ b/pkg/memory/sqlc/queries/rl.sql @@ -0,0 +1,202 @@ +-- RL (Reinforcement Learning) queries for Memelord integration +-- Task baseline queries for per-agent performance statistics + +-- name: GetTaskBaseline :one +-- Get the baseline statistics for an agent +SELECT agent_id, + count, + mean_tokens, + mean_errors, + mean_user_corrections, + m2_tokens, + m2_errors, + m2_user_corrections, + updated_at +FROM task_baselines +WHERE agent_id = sqlc.arg(agent_id) +LIMIT 1; + +-- name: UpdateTaskBaseline :exec +-- Insert or replace task baseline statistics for an agent +INSERT INTO task_baselines ( + agent_id, + count, + mean_tokens, + mean_errors, + mean_user_corrections, + m2_tokens, + m2_errors, + m2_user_corrections, + updated_at + ) +VALUES ( + sqlc.arg(agent_id), + sqlc.arg(count), + sqlc.arg(mean_tokens), + sqlc.arg(mean_errors), + sqlc.arg(mean_user_corrections), + sqlc.arg(m2_tokens), + sqlc.arg(m2_errors), + sqlc.arg(m2_user_corrections), + datetime('now') + ) +ON CONFLICT (agent_id) DO +UPDATE +SET count = excluded.count, + mean_tokens = excluded.mean_tokens, + mean_errors = excluded.mean_errors, + mean_user_corrections = excluded.mean_user_corrections, + m2_tokens = excluded.m2_tokens, + m2_errors = excluded.m2_errors, + m2_user_corrections = excluded.m2_user_corrections, + updated_at = excluded.updated_at; + +-- name: UpdateMemoryWeight :exec +-- Update the RL weight and credit for a specific memory item +UPDATE recall_items +SET rl_weight = sqlc.arg(rl_weight), + rl_credit = sqlc.arg(rl_credit), + updated_at = datetime('now') +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); + +-- name: UpdateMemorySelfReportScore :exec +-- Update the self-reported score for a memory item +UPDATE recall_items +SET self_report_score = sqlc.arg(self_report_score), + updated_at = datetime('now') +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); + +-- name: IncrementTaskRetrievalCount :exec +-- Increment the task retrieval counter for a memory item +UPDATE recall_items +SET task_retrieval_count = task_retrieval_count + 1, + updated_at = datetime('now') +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); + +-- name: GetMemoriesByRetrievalCount :many +-- Get memories ordered by their task retrieval count (for RL analysis) +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + rl_weight, + rl_credit, + self_report_score, + task_retrieval_count, + created_at, + updated_at +FROM recall_items +WHERE agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL +ORDER BY task_retrieval_count DESC +LIMIT sqlc.arg(lim); + +-- name: ListHighValueMemories :many +-- List memories with high RL weights (credits) for priority retention +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + rl_weight, + rl_credit, + self_report_score, + task_retrieval_count, + created_at, + updated_at +FROM recall_items +WHERE agent_id = sqlc.arg(agent_id) + AND suppressed_at IS NULL + AND ( + rl_credit > sqlc.arg(min_credit) + OR rl_weight > sqlc.arg(min_weight) + ) +ORDER BY rl_credit DESC NULLS LAST +LIMIT sqlc.arg(lim); + +-- name: StoreTaskCompletion :one +-- Store a task completion record for RL analysis +INSERT INTO task_completions ( + id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed + ) +VALUES ( + sqlc.arg(id), + sqlc.arg(agent_id), + sqlc.arg(conversation_id), + sqlc.arg(run_id), + sqlc.arg(description), + sqlc.arg(tokens_used), + sqlc.arg(tool_calls), + sqlc.arg(errors), + sqlc.arg(user_corrections), + sqlc.arg(completed) + ) +RETURNING id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed, + created_at; + +-- name: GetCompletedTasks :many +-- Get tasks completed since the given time for RL processing +SELECT id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed, + created_at +FROM task_completions +WHERE agent_id = sqlc.arg(agent_id) + AND created_at > sqlc.arg(since) + AND completed = 1 +ORDER BY created_at ASC; + +-- name: StoreTaskRetrieval :exec +-- Store a memory retrieval record for a task +INSERT INTO task_retrievals (id, task_id, memory_id, similarity) +VALUES (sqlc.arg(id), sqlc.arg(task_id), sqlc.arg(memory_id), sqlc.arg(similarity)) +ON CONFLICT (task_id, memory_id) DO UPDATE SET + similarity = excluded.similarity; + +-- name: GetRetrievedMemories :many +-- Get memories retrieved during a specific task with their self-report scores +SELECT tr.memory_id, + tr.similarity, + ri.self_report_score +FROM task_retrievals tr + JOIN recall_items ri ON tr.memory_id = ri.id +WHERE tr.task_id = sqlc.arg(task_id); diff --git a/pkg/memory/sqlc/queries/soft_delete.sql b/pkg/memory/sqlc/queries/soft_delete.sql new file mode 100644 index 000000000..2a0b1bb40 --- /dev/null +++ b/pkg/memory/sqlc/queries/soft_delete.sql @@ -0,0 +1,69 @@ +-- Soft delete queries (T2.4: quarantine before permanent deletion) +-- name: SoftDeleteRecallItem :exec +-- Sets suppressed_at for a recall item instead of permanently deleting it. +UPDATE recall_items +SET suppressed_at = datetime('now') +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); +-- name: SoftDeleteArchivalChunks :exec +-- Sets suppressed_at for all chunks belonging to a recall item. +UPDATE archival_chunks +SET suppressed_at = datetime('now') +WHERE recall_id = sqlc.arg(recall_id); +-- name: ListQuarantinedRecallItems :many +-- Returns recall items that have been soft-deleted (suppressed) and are +-- eligible for permanent deletion (older than quarantine period). +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + created_at, + updated_at, + suppressed_at +FROM recall_items +WHERE suppressed_at IS NOT NULL + AND suppressed_at < sqlc.arg(before_date) +ORDER BY suppressed_at ASC +LIMIT sqlc.arg(lim); +-- name: ListQuarantinedArchivalChunks :many +-- Returns archival chunks that have been soft-deleted and are +-- eligible for permanent deletion. +SELECT id, + recall_id, + chunk_index, + content, + embedding, + source, + hash, + created_at, + suppressed_at +FROM archival_chunks +WHERE suppressed_at IS NOT NULL + AND suppressed_at < sqlc.arg(before_date) +ORDER BY suppressed_at ASC +LIMIT sqlc.arg(lim); +-- name: HardDeleteRecallItem :exec +-- Permanently deletes a recall item (after quarantine period). +DELETE FROM recall_items +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); +-- name: HardDeleteArchivalChunks :exec +-- Permanently deletes archival chunks for a recall item. +DELETE FROM archival_chunks +WHERE recall_id = sqlc.arg(recall_id); +-- name: HardDeleteChunk :exec +-- Permanently deletes a single archival chunk by ID. +DELETE FROM archival_chunks +WHERE id = sqlc.arg(id); +-- name: RestoreRecallItem :exec +-- Restores a soft-deleted recall item by clearing suppressed_at. +UPDATE recall_items +SET suppressed_at = NULL +WHERE id = sqlc.arg(id) + AND agent_id = sqlc.arg(agent_id); \ No newline at end of file diff --git a/pkg/memory/sqlc/recall.sql.go b/pkg/memory/sqlc/recall.sql.go index 946f9968b..56d272dea 100644 --- a/pkg/memory/sqlc/recall.sql.go +++ b/pkg/memory/sqlc/recall.sql.go @@ -8,6 +8,7 @@ package sqlc import ( "context" "strings" + "time" "github.com/ZanzyTHEbar/dragonscale/pkg/ids" "github.com/ZanzyTHEbar/dragonscale/pkg/memory" @@ -17,6 +18,7 @@ const CountRecallItems = `-- name: CountRecallItems :one SELECT COUNT(*) FROM recall_items WHERE agent_id = ?1 + AND suppressed_at IS NULL AND ( session_key = ?2 OR ?2 = '' @@ -33,6 +35,7 @@ type CountRecallItemsParams struct { // SELECT COUNT(*) // FROM recall_items // WHERE agent_id = ?1 +// AND suppressed_at IS NULL // AND ( // session_key = ?2 // OR ?2 = '' @@ -108,6 +111,7 @@ SELECT id, FROM recall_items WHERE id = ?1 AND agent_id = ?2 + AND suppressed_at IS NULL LIMIT 1 ` @@ -116,6 +120,21 @@ type GetRecallItemParams struct { AgentID string `db:"agent_id" json:"agent_id"` } +type GetRecallItemRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // GetRecallItem // // SELECT id, @@ -133,10 +152,11 @@ type GetRecallItemParams struct { // FROM recall_items // WHERE id = ?1 // AND agent_id = ?2 +// AND suppressed_at IS NULL // LIMIT 1 -func (q *Queries) GetRecallItem(ctx context.Context, arg GetRecallItemParams) (RecallItem, error) { +func (q *Queries) GetRecallItem(ctx context.Context, arg GetRecallItemParams) (GetRecallItemRow, error) { row := q.db.QueryRowContext(ctx, GetRecallItem, arg.ID, arg.AgentID) - var i RecallItem + var i GetRecallItemRow err := row.Scan( &i.ID, &i.AgentID, @@ -170,6 +190,7 @@ SELECT id, FROM recall_items WHERE id IN (/*SLICE:ids*/?) AND agent_id = ?2 + AND suppressed_at IS NULL ` type GetRecallItemsByIDsParams struct { @@ -177,6 +198,21 @@ type GetRecallItemsByIDsParams struct { AgentID string `db:"agent_id" json:"agent_id"` } +type GetRecallItemsByIDsRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // GetRecallItemsByIDs // // SELECT id, @@ -194,7 +230,8 @@ type GetRecallItemsByIDsParams struct { // FROM recall_items // WHERE id IN (/*SLICE:ids*/?) // AND agent_id = ?2 -func (q *Queries) GetRecallItemsByIDs(ctx context.Context, arg GetRecallItemsByIDsParams) ([]RecallItem, error) { +// AND suppressed_at IS NULL +func (q *Queries) GetRecallItemsByIDs(ctx context.Context, arg GetRecallItemsByIDsParams) ([]GetRecallItemsByIDsRow, error) { query := GetRecallItemsByIDs var queryParams []interface{} if len(arg.Ids) > 0 { @@ -211,9 +248,9 @@ func (q *Queries) GetRecallItemsByIDs(ctx context.Context, arg GetRecallItemsByI return nil, err } defer rows.Close() - items := []RecallItem{} + items := []GetRecallItemsByIDsRow{} for rows.Next() { - var i RecallItem + var i GetRecallItemsByIDsRow if err := rows.Scan( &i.ID, &i.AgentID, @@ -253,6 +290,7 @@ INSERT INTO recall_items ( decay_rate, content, tags, + rl_weight, created_at, updated_at ) @@ -267,6 +305,7 @@ VALUES ( ?8, ?9, ?10, + ?11, datetime('now'), datetime('now') ) @@ -280,6 +319,7 @@ RETURNING id, decay_rate, content, tags, + rl_weight, created_at, updated_at ` @@ -295,6 +335,23 @@ type InsertRecallItemParams struct { DecayRate float64 `db:"decay_rate" json:"decay_rate"` Content string `db:"content" json:"content"` Tags string `db:"tags" json:"tags"` + RlWeight *float64 `db:"rl_weight" json:"rl_weight"` +} + +type InsertRecallItemRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + RlWeight *float64 `db:"rl_weight" json:"rl_weight"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` } // Recall Item queries @@ -310,6 +367,7 @@ type InsertRecallItemParams struct { // decay_rate, // content, // tags, +// rl_weight, // created_at, // updated_at // ) @@ -324,6 +382,7 @@ type InsertRecallItemParams struct { // ?8, // ?9, // ?10, +// ?11, // datetime('now'), // datetime('now') // ) @@ -337,9 +396,10 @@ type InsertRecallItemParams struct { // decay_rate, // content, // tags, +// rl_weight, // created_at, // updated_at -func (q *Queries) InsertRecallItem(ctx context.Context, arg InsertRecallItemParams) (RecallItem, error) { +func (q *Queries) InsertRecallItem(ctx context.Context, arg InsertRecallItemParams) (InsertRecallItemRow, error) { row := q.db.QueryRowContext(ctx, InsertRecallItem, arg.ID, arg.AgentID, @@ -351,8 +411,9 @@ func (q *Queries) InsertRecallItem(ctx context.Context, arg InsertRecallItemPara arg.DecayRate, arg.Content, arg.Tags, + arg.RlWeight, ) - var i RecallItem + var i InsertRecallItemRow err := row.Scan( &i.ID, &i.AgentID, @@ -364,6 +425,7 @@ func (q *Queries) InsertRecallItem(ctx context.Context, arg InsertRecallItemPara &i.DecayRate, &i.Content, &i.Tags, + &i.RlWeight, &i.CreatedAt, &i.UpdatedAt, ) @@ -421,6 +483,21 @@ type InsertSessionMessageParams struct { Content string `db:"content" json:"content"` } +type InsertSessionMessageRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // InsertSessionMessage // // INSERT INTO recall_items ( @@ -463,7 +540,7 @@ type InsertSessionMessageParams struct { // tags, // created_at, // updated_at -func (q *Queries) InsertSessionMessage(ctx context.Context, arg InsertSessionMessageParams) (RecallItem, error) { +func (q *Queries) InsertSessionMessage(ctx context.Context, arg InsertSessionMessageParams) (InsertSessionMessageRow, error) { row := q.db.QueryRowContext(ctx, InsertSessionMessage, arg.ID, arg.AgentID, @@ -471,7 +548,7 @@ func (q *Queries) InsertSessionMessage(ctx context.Context, arg InsertSessionMes arg.Role, arg.Content, ) - var i RecallItem + var i InsertSessionMessageRow err := row.Scan( &i.ID, &i.AgentID, @@ -504,6 +581,7 @@ SELECT id, updated_at FROM recall_items WHERE agent_id = ?1 + AND suppressed_at IS NULL AND ( session_key = ?2 OR ?2 = '' @@ -519,6 +597,21 @@ type ListRecallItemsParams struct { Lim int64 `db:"lim" json:"lim"` } +type ListRecallItemsRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // ListRecallItems // // SELECT id, @@ -535,13 +628,14 @@ type ListRecallItemsParams struct { // updated_at // FROM recall_items // WHERE agent_id = ?1 +// AND suppressed_at IS NULL // AND ( // session_key = ?2 // OR ?2 = '' // ) // ORDER BY created_at DESC // LIMIT ?4 OFFSET ?3 -func (q *Queries) ListRecallItems(ctx context.Context, arg ListRecallItemsParams) ([]RecallItem, error) { +func (q *Queries) ListRecallItems(ctx context.Context, arg ListRecallItemsParams) ([]ListRecallItemsRow, error) { rows, err := q.db.QueryContext(ctx, ListRecallItems, arg.AgentID, arg.SessionKey, @@ -552,9 +646,9 @@ func (q *Queries) ListRecallItems(ctx context.Context, arg ListRecallItemsParams return nil, err } defer rows.Close() - items := []RecallItem{} + items := []ListRecallItemsRow{} for rows.Next() { - var i RecallItem + var i ListRecallItemsRow if err := rows.Scan( &i.ID, &i.AgentID, @@ -614,6 +708,21 @@ type ListSessionMessagesParams struct { Lim int64 `db:"lim" json:"lim"` } +type ListSessionMessagesRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // ListSessionMessages // // SELECT id, @@ -638,7 +747,7 @@ type ListSessionMessagesParams struct { // ) // ORDER BY created_at ASC // LIMIT ?4 -func (q *Queries) ListSessionMessages(ctx context.Context, arg ListSessionMessagesParams) ([]RecallItem, error) { +func (q *Queries) ListSessionMessages(ctx context.Context, arg ListSessionMessagesParams) ([]ListSessionMessagesRow, error) { rows, err := q.db.QueryContext(ctx, ListSessionMessages, arg.AgentID, arg.SessionKey, @@ -649,9 +758,9 @@ func (q *Queries) ListSessionMessages(ctx context.Context, arg ListSessionMessag return nil, err } defer rows.Close() - items := []RecallItem{} + items := []ListSessionMessagesRow{} for rows.Next() { - var i RecallItem + var i ListSessionMessagesRow if err := rows.Scan( &i.ID, &i.AgentID, @@ -712,6 +821,21 @@ type ListSessionMessagesPagedParams struct { Lim int64 `db:"lim" json:"lim"` } +type ListSessionMessagesPagedRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // ListSessionMessagesPaged // // SELECT id, @@ -736,7 +860,7 @@ type ListSessionMessagesPagedParams struct { // ) // ORDER BY created_at ASC // LIMIT ?5 OFFSET ?4 -func (q *Queries) ListSessionMessagesPaged(ctx context.Context, arg ListSessionMessagesPagedParams) ([]RecallItem, error) { +func (q *Queries) ListSessionMessagesPaged(ctx context.Context, arg ListSessionMessagesPagedParams) ([]ListSessionMessagesPagedRow, error) { rows, err := q.db.QueryContext(ctx, ListSessionMessagesPaged, arg.AgentID, arg.SessionKey, @@ -748,9 +872,9 @@ func (q *Queries) ListSessionMessagesPaged(ctx context.Context, arg ListSessionM return nil, err } defer rows.Close() - items := []RecallItem{} + items := []ListSessionMessagesPagedRow{} for rows.Next() { - var i RecallItem + var i ListSessionMessagesPagedRow if err := rows.Scan( &i.ID, &i.AgentID, @@ -794,6 +918,7 @@ SELECT ri.id, FROM recall_items ri WHERE ri.content LIKE '%' || ?1 || '%' AND ri.agent_id = ?2 + AND ri.suppressed_at IS NULL ORDER BY ri.importance DESC LIMIT ?3 ` @@ -804,6 +929,21 @@ type SearchRecallByKeywordParams struct { Lim int64 `db:"lim" json:"lim"` } +type SearchRecallByKeywordRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + // SearchRecallByKeyword // // SELECT ri.id, @@ -821,17 +961,18 @@ type SearchRecallByKeywordParams struct { // FROM recall_items ri // WHERE ri.content LIKE '%' || ?1 || '%' // AND ri.agent_id = ?2 +// AND ri.suppressed_at IS NULL // ORDER BY ri.importance DESC // LIMIT ?3 -func (q *Queries) SearchRecallByKeyword(ctx context.Context, arg SearchRecallByKeywordParams) ([]RecallItem, error) { +func (q *Queries) SearchRecallByKeyword(ctx context.Context, arg SearchRecallByKeywordParams) ([]SearchRecallByKeywordRow, error) { rows, err := q.db.QueryContext(ctx, SearchRecallByKeyword, arg.Keyword, arg.AgentID, arg.Lim) if err != nil { return nil, err } defer rows.Close() - items := []RecallItem{} + items := []SearchRecallByKeywordRow{} for rows.Next() { - var i RecallItem + var i SearchRecallByKeywordRow if err := rows.Scan( &i.ID, &i.AgentID, diff --git a/pkg/memory/sqlc/recall_decay.sql.go b/pkg/memory/sqlc/recall_decay.sql.go new file mode 100644 index 000000000..328e49b4e --- /dev/null +++ b/pkg/memory/sqlc/recall_decay.sql.go @@ -0,0 +1,175 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: recall_decay.sql + +package sqlc + +import ( + "context" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" +) + +const CountArchivalChunksWithoutEmbedding = `-- name: CountArchivalChunksWithoutEmbedding :one +SELECT COUNT(*) +FROM archival_chunks +WHERE embedding IS NULL + AND suppressed_at IS NULL +` + +// CountArchivalChunksWithoutEmbedding +// +// SELECT COUNT(*) +// FROM archival_chunks +// WHERE embedding IS NULL +// AND suppressed_at IS NULL +func (q *Queries) CountArchivalChunksWithoutEmbedding(ctx context.Context) (int64, error) { + row := q.db.QueryRowContext(ctx, CountArchivalChunksWithoutEmbedding) + var count int64 + err := row.Scan(&count) + return count, err +} + +const DecayRecallImportanceBatch = `-- name: DecayRecallImportanceBatch :exec +UPDATE recall_items +SET importance = MAX( + recall_items.importance * ?1, + ?2 + ), + updated_at = datetime('now') +WHERE recall_items.id IN ( + SELECT ri.id + FROM recall_items ri + WHERE ri.importance > ?2 + AND ri.suppressed_at IS NULL + ORDER BY ri.updated_at ASC + LIMIT ?3 + ) +` + +type DecayRecallImportanceBatchParams struct { + Factor float64 `db:"factor" json:"factor"` + FloorVal interface{} `db:"floor_val" json:"floor_val"` + BatchSize int64 `db:"batch_size" json:"batch_size"` +} + +// Batch decay of recall item importance (Cortex decay task) +// +// UPDATE recall_items +// SET importance = MAX( +// recall_items.importance * ?1, +// ?2 +// ), +// updated_at = datetime('now') +// WHERE recall_items.id IN ( +// SELECT ri.id +// FROM recall_items ri +// WHERE ri.importance > ?2 +// AND ri.suppressed_at IS NULL +// ORDER BY ri.updated_at ASC +// LIMIT ?3 +// ) +func (q *Queries) DecayRecallImportanceBatch(ctx context.Context, arg DecayRecallImportanceBatchParams) error { + _, err := q.db.ExecContext(ctx, DecayRecallImportanceBatch, arg.Factor, arg.FloorVal, arg.BatchSize) + return err +} + +const ListArchivalChunksWithoutEmbedding = `-- name: ListArchivalChunksWithoutEmbedding :many +SELECT id, + recall_id, + chunk_index, + content, + embedding, + source, + hash, + created_at +FROM archival_chunks +WHERE embedding IS NULL + AND suppressed_at IS NULL +LIMIT ?1 +` + +type ListArchivalChunksWithoutEmbeddingParams struct { + Lim int64 `db:"lim" json:"lim"` +} + +type ListArchivalChunksWithoutEmbeddingRow struct { + ID ids.UUID `db:"id" json:"id"` + RecallID ids.UUID `db:"recall_id" json:"recall_id"` + ChunkIndex int64 `db:"chunk_index" json:"chunk_index"` + Content string `db:"content" json:"content"` + Embedding memory.Embedding `db:"embedding" json:"embedding"` + Source string `db:"source" json:"source"` + Hash string `db:"hash" json:"hash"` + CreatedAt time.Time `db:"created_at" json:"created_at"` +} + +// ListArchivalChunksWithoutEmbedding +// +// SELECT id, +// recall_id, +// chunk_index, +// content, +// embedding, +// source, +// hash, +// created_at +// FROM archival_chunks +// WHERE embedding IS NULL +// AND suppressed_at IS NULL +// LIMIT ?1 +func (q *Queries) ListArchivalChunksWithoutEmbedding(ctx context.Context, arg ListArchivalChunksWithoutEmbeddingParams) ([]ListArchivalChunksWithoutEmbeddingRow, error) { + rows, err := q.db.QueryContext(ctx, ListArchivalChunksWithoutEmbedding, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListArchivalChunksWithoutEmbeddingRow{} + for rows.Next() { + var i ListArchivalChunksWithoutEmbeddingRow + if err := rows.Scan( + &i.ID, + &i.RecallID, + &i.ChunkIndex, + &i.Content, + &i.Embedding, + &i.Source, + &i.Hash, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const UpdateArchivalChunkEmbedding = `-- name: UpdateArchivalChunkEmbedding :exec +UPDATE archival_chunks +SET embedding = ?1 +WHERE id = ?2 +` + +type UpdateArchivalChunkEmbeddingParams struct { + Embedding memory.Embedding `db:"embedding" json:"embedding"` + ID ids.UUID `db:"id" json:"id"` +} + +// UpdateArchivalChunkEmbedding +// +// UPDATE archival_chunks +// SET embedding = ?1 +// WHERE id = ?2 +func (q *Queries) UpdateArchivalChunkEmbedding(ctx context.Context, arg UpdateArchivalChunkEmbeddingParams) error { + _, err := q.db.ExecContext(ctx, UpdateArchivalChunkEmbedding, arg.Embedding, arg.ID) + return err +} diff --git a/pkg/memory/sqlc/rl.sql.go b/pkg/memory/sqlc/rl.sql.go new file mode 100644 index 000000000..21a446808 --- /dev/null +++ b/pkg/memory/sqlc/rl.sql.go @@ -0,0 +1,758 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: rl.sql + +package sqlc + +import ( + "context" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" +) + +const GetCompletedTasks = `-- name: GetCompletedTasks :many +SELECT id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed, + created_at +FROM task_completions +WHERE agent_id = ?1 + AND created_at > ?2 + AND completed = 1 +ORDER BY created_at ASC +` + +type GetCompletedTasksParams struct { + AgentID string `db:"agent_id" json:"agent_id"` + Since time.Time `db:"since" json:"since"` +} + +// Get tasks completed since the given time for RL processing +// +// SELECT id, +// agent_id, +// conversation_id, +// run_id, +// description, +// tokens_used, +// tool_calls, +// errors, +// user_corrections, +// completed, +// created_at +// FROM task_completions +// WHERE agent_id = ?1 +// AND created_at > ?2 +// AND completed = 1 +// ORDER BY created_at ASC +func (q *Queries) GetCompletedTasks(ctx context.Context, arg GetCompletedTasksParams) ([]TaskCompletion, error) { + rows, err := q.db.QueryContext(ctx, GetCompletedTasks, arg.AgentID, arg.Since) + if err != nil { + return nil, err + } + defer rows.Close() + items := []TaskCompletion{} + for rows.Next() { + var i TaskCompletion + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.ConversationID, + &i.RunID, + &i.Description, + &i.TokensUsed, + &i.ToolCalls, + &i.Errors, + &i.UserCorrections, + &i.Completed, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const GetMemoriesByRetrievalCount = `-- name: GetMemoriesByRetrievalCount :many +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + rl_weight, + rl_credit, + self_report_score, + task_retrieval_count, + created_at, + updated_at +FROM recall_items +WHERE agent_id = ?1 + AND suppressed_at IS NULL +ORDER BY task_retrieval_count DESC +LIMIT ?2 +` + +type GetMemoriesByRetrievalCountParams struct { + AgentID string `db:"agent_id" json:"agent_id"` + Lim int64 `db:"lim" json:"lim"` +} + +type GetMemoriesByRetrievalCountRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + RlWeight *float64 `db:"rl_weight" json:"rl_weight"` + RlCredit *float64 `db:"rl_credit" json:"rl_credit"` + SelfReportScore *int64 `db:"self_report_score" json:"self_report_score"` + TaskRetrievalCount *int64 `db:"task_retrieval_count" json:"task_retrieval_count"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +// Get memories ordered by their task retrieval count (for RL analysis) +// +// SELECT id, +// agent_id, +// session_key, +// role, +// sector, +// importance, +// salience, +// decay_rate, +// content, +// tags, +// rl_weight, +// rl_credit, +// self_report_score, +// task_retrieval_count, +// created_at, +// updated_at +// FROM recall_items +// WHERE agent_id = ?1 +// AND suppressed_at IS NULL +// ORDER BY task_retrieval_count DESC +// LIMIT ?2 +func (q *Queries) GetMemoriesByRetrievalCount(ctx context.Context, arg GetMemoriesByRetrievalCountParams) ([]GetMemoriesByRetrievalCountRow, error) { + rows, err := q.db.QueryContext(ctx, GetMemoriesByRetrievalCount, arg.AgentID, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetMemoriesByRetrievalCountRow{} + for rows.Next() { + var i GetMemoriesByRetrievalCountRow + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.SessionKey, + &i.Role, + &i.Sector, + &i.Importance, + &i.Salience, + &i.DecayRate, + &i.Content, + &i.Tags, + &i.RlWeight, + &i.RlCredit, + &i.SelfReportScore, + &i.TaskRetrievalCount, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const GetRetrievedMemories = `-- name: GetRetrievedMemories :many +SELECT tr.memory_id, + tr.similarity, + ri.self_report_score +FROM task_retrievals tr + JOIN recall_items ri ON tr.memory_id = ri.id +WHERE tr.task_id = ?1 +` + +type GetRetrievedMemoriesParams struct { + TaskID ids.UUID `db:"task_id" json:"task_id"` +} + +type GetRetrievedMemoriesRow struct { + MemoryID ids.UUID `db:"memory_id" json:"memory_id"` + Similarity float64 `db:"similarity" json:"similarity"` + SelfReportScore *int64 `db:"self_report_score" json:"self_report_score"` +} + +// Get memories retrieved during a specific task with their self-report scores +// +// SELECT tr.memory_id, +// tr.similarity, +// ri.self_report_score +// FROM task_retrievals tr +// JOIN recall_items ri ON tr.memory_id = ri.id +// WHERE tr.task_id = ?1 +func (q *Queries) GetRetrievedMemories(ctx context.Context, arg GetRetrievedMemoriesParams) ([]GetRetrievedMemoriesRow, error) { + rows, err := q.db.QueryContext(ctx, GetRetrievedMemories, arg.TaskID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetRetrievedMemoriesRow{} + for rows.Next() { + var i GetRetrievedMemoriesRow + if err := rows.Scan(&i.MemoryID, &i.Similarity, &i.SelfReportScore); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const GetTaskBaseline = `-- name: GetTaskBaseline :one + +SELECT agent_id, + count, + mean_tokens, + mean_errors, + mean_user_corrections, + m2_tokens, + m2_errors, + m2_user_corrections, + updated_at +FROM task_baselines +WHERE agent_id = ?1 +LIMIT 1 +` + +type GetTaskBaselineParams struct { + AgentID string `db:"agent_id" json:"agent_id"` +} + +// RL (Reinforcement Learning) queries for Memelord integration +// Task baseline queries for per-agent performance statistics +// Get the baseline statistics for an agent +// +// SELECT agent_id, +// count, +// mean_tokens, +// mean_errors, +// mean_user_corrections, +// m2_tokens, +// m2_errors, +// m2_user_corrections, +// updated_at +// FROM task_baselines +// WHERE agent_id = ?1 +// LIMIT 1 +func (q *Queries) GetTaskBaseline(ctx context.Context, arg GetTaskBaselineParams) (TaskBaseline, error) { + row := q.db.QueryRowContext(ctx, GetTaskBaseline, arg.AgentID) + var i TaskBaseline + err := row.Scan( + &i.AgentID, + &i.Count, + &i.MeanTokens, + &i.MeanErrors, + &i.MeanUserCorrections, + &i.M2Tokens, + &i.M2Errors, + &i.M2UserCorrections, + &i.UpdatedAt, + ) + return i, err +} + +const IncrementTaskRetrievalCount = `-- name: IncrementTaskRetrievalCount :exec +UPDATE recall_items +SET task_retrieval_count = task_retrieval_count + 1, + updated_at = datetime('now') +WHERE id = ?1 + AND agent_id = ?2 +` + +type IncrementTaskRetrievalCountParams struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Increment the task retrieval counter for a memory item +// +// UPDATE recall_items +// SET task_retrieval_count = task_retrieval_count + 1, +// updated_at = datetime('now') +// WHERE id = ?1 +// AND agent_id = ?2 +func (q *Queries) IncrementTaskRetrievalCount(ctx context.Context, arg IncrementTaskRetrievalCountParams) error { + _, err := q.db.ExecContext(ctx, IncrementTaskRetrievalCount, arg.ID, arg.AgentID) + return err +} + +const ListHighValueMemories = `-- name: ListHighValueMemories :many +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + rl_weight, + rl_credit, + self_report_score, + task_retrieval_count, + created_at, + updated_at +FROM recall_items +WHERE agent_id = ?1 + AND suppressed_at IS NULL + AND ( + rl_credit > ?2 + OR rl_weight > ?3 + ) +ORDER BY rl_credit DESC NULLS LAST +LIMIT ?4 +` + +type ListHighValueMemoriesParams struct { + AgentID string `db:"agent_id" json:"agent_id"` + MinCredit *float64 `db:"min_credit" json:"min_credit"` + MinWeight *float64 `db:"min_weight" json:"min_weight"` + Lim int64 `db:"lim" json:"lim"` +} + +type ListHighValueMemoriesRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + RlWeight *float64 `db:"rl_weight" json:"rl_weight"` + RlCredit *float64 `db:"rl_credit" json:"rl_credit"` + SelfReportScore *int64 `db:"self_report_score" json:"self_report_score"` + TaskRetrievalCount *int64 `db:"task_retrieval_count" json:"task_retrieval_count"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` +} + +// List memories with high RL weights (credits) for priority retention +// +// SELECT id, +// agent_id, +// session_key, +// role, +// sector, +// importance, +// salience, +// decay_rate, +// content, +// tags, +// rl_weight, +// rl_credit, +// self_report_score, +// task_retrieval_count, +// created_at, +// updated_at +// FROM recall_items +// WHERE agent_id = ?1 +// AND suppressed_at IS NULL +// AND ( +// rl_credit > ?2 +// OR rl_weight > ?3 +// ) +// ORDER BY rl_credit DESC NULLS LAST +// LIMIT ?4 +func (q *Queries) ListHighValueMemories(ctx context.Context, arg ListHighValueMemoriesParams) ([]ListHighValueMemoriesRow, error) { + rows, err := q.db.QueryContext(ctx, ListHighValueMemories, + arg.AgentID, + arg.MinCredit, + arg.MinWeight, + arg.Lim, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListHighValueMemoriesRow{} + for rows.Next() { + var i ListHighValueMemoriesRow + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.SessionKey, + &i.Role, + &i.Sector, + &i.Importance, + &i.Salience, + &i.DecayRate, + &i.Content, + &i.Tags, + &i.RlWeight, + &i.RlCredit, + &i.SelfReportScore, + &i.TaskRetrievalCount, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const StoreTaskCompletion = `-- name: StoreTaskCompletion :one +INSERT INTO task_completions ( + id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed + ) +VALUES ( + ?1, + ?2, + ?3, + ?4, + ?5, + ?6, + ?7, + ?8, + ?9, + ?10 + ) +RETURNING id, + agent_id, + conversation_id, + run_id, + description, + tokens_used, + tool_calls, + errors, + user_corrections, + completed, + created_at +` + +type StoreTaskCompletionParams struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + ConversationID ids.UUID `db:"conversation_id" json:"conversation_id"` + RunID ids.UUID `db:"run_id" json:"run_id"` + Description string `db:"description" json:"description"` + TokensUsed *int64 `db:"tokens_used" json:"tokens_used"` + ToolCalls *int64 `db:"tool_calls" json:"tool_calls"` + Errors *int64 `db:"errors" json:"errors"` + UserCorrections *int64 `db:"user_corrections" json:"user_corrections"` + Completed bool `db:"completed" json:"completed"` +} + +// Store a task completion record for RL analysis +// +// INSERT INTO task_completions ( +// id, +// agent_id, +// conversation_id, +// run_id, +// description, +// tokens_used, +// tool_calls, +// errors, +// user_corrections, +// completed +// ) +// VALUES ( +// ?1, +// ?2, +// ?3, +// ?4, +// ?5, +// ?6, +// ?7, +// ?8, +// ?9, +// ?10 +// ) +// RETURNING id, +// agent_id, +// conversation_id, +// run_id, +// description, +// tokens_used, +// tool_calls, +// errors, +// user_corrections, +// completed, +// created_at +func (q *Queries) StoreTaskCompletion(ctx context.Context, arg StoreTaskCompletionParams) (TaskCompletion, error) { + row := q.db.QueryRowContext(ctx, StoreTaskCompletion, + arg.ID, + arg.AgentID, + arg.ConversationID, + arg.RunID, + arg.Description, + arg.TokensUsed, + arg.ToolCalls, + arg.Errors, + arg.UserCorrections, + arg.Completed, + ) + var i TaskCompletion + err := row.Scan( + &i.ID, + &i.AgentID, + &i.ConversationID, + &i.RunID, + &i.Description, + &i.TokensUsed, + &i.ToolCalls, + &i.Errors, + &i.UserCorrections, + &i.Completed, + &i.CreatedAt, + ) + return i, err +} + +const StoreTaskRetrieval = `-- name: StoreTaskRetrieval :exec +INSERT INTO task_retrievals (id, task_id, memory_id, similarity) +VALUES (?1, ?2, ?3, ?4) +ON CONFLICT (task_id, memory_id) DO UPDATE SET + similarity = excluded.similarity +` + +type StoreTaskRetrievalParams struct { + ID ids.UUID `db:"id" json:"id"` + TaskID ids.UUID `db:"task_id" json:"task_id"` + MemoryID ids.UUID `db:"memory_id" json:"memory_id"` + Similarity float64 `db:"similarity" json:"similarity"` +} + +// Store a memory retrieval record for a task +// +// INSERT INTO task_retrievals (id, task_id, memory_id, similarity) +// VALUES (?1, ?2, ?3, ?4) +// ON CONFLICT (task_id, memory_id) DO UPDATE SET +// similarity = excluded.similarity +func (q *Queries) StoreTaskRetrieval(ctx context.Context, arg StoreTaskRetrievalParams) error { + _, err := q.db.ExecContext(ctx, StoreTaskRetrieval, + arg.ID, + arg.TaskID, + arg.MemoryID, + arg.Similarity, + ) + return err +} + +const UpdateMemorySelfReportScore = `-- name: UpdateMemorySelfReportScore :exec +UPDATE recall_items +SET self_report_score = ?1, + updated_at = datetime('now') +WHERE id = ?2 + AND agent_id = ?3 +` + +type UpdateMemorySelfReportScoreParams struct { + SelfReportScore *int64 `db:"self_report_score" json:"self_report_score"` + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Update the self-reported score for a memory item +// +// UPDATE recall_items +// SET self_report_score = ?1, +// updated_at = datetime('now') +// WHERE id = ?2 +// AND agent_id = ?3 +func (q *Queries) UpdateMemorySelfReportScore(ctx context.Context, arg UpdateMemorySelfReportScoreParams) error { + _, err := q.db.ExecContext(ctx, UpdateMemorySelfReportScore, arg.SelfReportScore, arg.ID, arg.AgentID) + return err +} + +const UpdateMemoryWeight = `-- name: UpdateMemoryWeight :exec +UPDATE recall_items +SET rl_weight = ?1, + rl_credit = ?2, + updated_at = datetime('now') +WHERE id = ?3 + AND agent_id = ?4 +` + +type UpdateMemoryWeightParams struct { + RlWeight *float64 `db:"rl_weight" json:"rl_weight"` + RlCredit *float64 `db:"rl_credit" json:"rl_credit"` + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Update the RL weight and credit for a specific memory item +// +// UPDATE recall_items +// SET rl_weight = ?1, +// rl_credit = ?2, +// updated_at = datetime('now') +// WHERE id = ?3 +// AND agent_id = ?4 +func (q *Queries) UpdateMemoryWeight(ctx context.Context, arg UpdateMemoryWeightParams) error { + _, err := q.db.ExecContext(ctx, UpdateMemoryWeight, + arg.RlWeight, + arg.RlCredit, + arg.ID, + arg.AgentID, + ) + return err +} + +const UpdateTaskBaseline = `-- name: UpdateTaskBaseline :exec +INSERT INTO task_baselines ( + agent_id, + count, + mean_tokens, + mean_errors, + mean_user_corrections, + m2_tokens, + m2_errors, + m2_user_corrections, + updated_at + ) +VALUES ( + ?1, + ?2, + ?3, + ?4, + ?5, + ?6, + ?7, + ?8, + datetime('now') + ) +ON CONFLICT (agent_id) DO +UPDATE +SET count = excluded.count, + mean_tokens = excluded.mean_tokens, + mean_errors = excluded.mean_errors, + mean_user_corrections = excluded.mean_user_corrections, + m2_tokens = excluded.m2_tokens, + m2_errors = excluded.m2_errors, + m2_user_corrections = excluded.m2_user_corrections, + updated_at = excluded.updated_at +` + +type UpdateTaskBaselineParams struct { + AgentID string `db:"agent_id" json:"agent_id"` + Count *int64 `db:"count" json:"count"` + MeanTokens *int64 `db:"mean_tokens" json:"mean_tokens"` + MeanErrors *float64 `db:"mean_errors" json:"mean_errors"` + MeanUserCorrections *float64 `db:"mean_user_corrections" json:"mean_user_corrections"` + M2Tokens *float64 `db:"m2_tokens" json:"m2_tokens"` + M2Errors *float64 `db:"m2_errors" json:"m2_errors"` + M2UserCorrections *float64 `db:"m2_user_corrections" json:"m2_user_corrections"` +} + +// Insert or replace task baseline statistics for an agent +// +// INSERT INTO task_baselines ( +// agent_id, +// count, +// mean_tokens, +// mean_errors, +// mean_user_corrections, +// m2_tokens, +// m2_errors, +// m2_user_corrections, +// updated_at +// ) +// VALUES ( +// ?1, +// ?2, +// ?3, +// ?4, +// ?5, +// ?6, +// ?7, +// ?8, +// datetime('now') +// ) +// ON CONFLICT (agent_id) DO +// UPDATE +// SET count = excluded.count, +// mean_tokens = excluded.mean_tokens, +// mean_errors = excluded.mean_errors, +// mean_user_corrections = excluded.mean_user_corrections, +// m2_tokens = excluded.m2_tokens, +// m2_errors = excluded.m2_errors, +// m2_user_corrections = excluded.m2_user_corrections, +// updated_at = excluded.updated_at +func (q *Queries) UpdateTaskBaseline(ctx context.Context, arg UpdateTaskBaselineParams) error { + _, err := q.db.ExecContext(ctx, UpdateTaskBaseline, + arg.AgentID, + arg.Count, + arg.MeanTokens, + arg.MeanErrors, + arg.MeanUserCorrections, + arg.M2Tokens, + arg.M2Errors, + arg.M2UserCorrections, + ) + return err +} diff --git a/pkg/memory/sqlc/schema.sql b/pkg/memory/sqlc/schema.sql index ee259f9dc..51f4273dd 100644 --- a/pkg/memory/sqlc/schema.sql +++ b/pkg/memory/sqlc/schema.sql @@ -26,7 +26,13 @@ CREATE TABLE IF NOT EXISTS recall_items ( content TEXT NOT NULL, tags TEXT NOT NULL DEFAULT '', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + suppressed_at DATETIME, -- soft delete timestamp (T2.4) + -- RL (Memelord) support columns + rl_weight REAL DEFAULT 1.0, -- current weight for credit assignment + rl_credit REAL, -- accumulated credit for this memory + self_report_score INTEGER, -- self-reported usefulness score + task_retrieval_count INTEGER DEFAULT 0 -- how many times retrieved for tasks ); CREATE INDEX IF NOT EXISTS idx_recall_agent_session ON recall_items(agent_id, session_key); CREATE INDEX IF NOT EXISTS idx_recall_sector ON recall_items(sector); @@ -44,7 +50,8 @@ CREATE TABLE IF NOT EXISTS archival_chunks ( embedding BLOB, source TEXT NOT NULL DEFAULT '', hash TEXT NOT NULL DEFAULT '', - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + suppressed_at DATETIME -- soft delete timestamp (T2.4) ); CREATE INDEX IF NOT EXISTS idx_chunks_recall ON archival_chunks(recall_id); CREATE INDEX IF NOT EXISTS idx_chunks_source ON archival_chunks(source); @@ -384,4 +391,79 @@ CREATE TABLE IF NOT EXISTS dag_edges ( ); CREATE INDEX IF NOT EXISTS idx_dag_edges_snapshot ON dag_edges(snapshot_id); CREATE INDEX IF NOT EXISTS idx_dag_edges_parent ON dag_edges(parent_node_id); -CREATE INDEX IF NOT EXISTS idx_dag_edges_child ON dag_edges(child_node_id); \ No newline at end of file +CREATE INDEX IF NOT EXISTS idx_dag_edges_child ON dag_edges(child_node_id); +-- ============================================================================ +-- Immutable Message Store (LCM ADR-001) +-- ============================================================================ +-- Append-only verbatim record of every message. Never modified or deleted. +CREATE TABLE IF NOT EXISTS immutable_messages ( + id BLOB PRIMARY KEY, + session_key TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + tool_call_id TEXT NOT NULL DEFAULT '', + tool_calls TEXT NOT NULL DEFAULT '', + token_estimate INTEGER NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); +CREATE INDEX IF NOT EXISTS idx_immutable_session_created ON immutable_messages(session_key, created_at); +-- ============================================================================ +-- Memory Graph Edges (ADR-004) +-- ============================================================================ +-- Typed, weighted relationships between memory items. +CREATE TABLE IF NOT EXISTS memory_edges ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + from_id BLOB NOT NULL, + to_id BLOB NOT NULL, + edge_type TEXT NOT NULL DEFAULT 'related_to', + weight REAL NOT NULL DEFAULT 1.0, + created_at DATETIME NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); +CREATE INDEX IF NOT EXISTS idx_memory_edges_from ON memory_edges(from_id); +CREATE INDEX IF NOT EXISTS idx_memory_edges_to ON memory_edges(to_id); +CREATE INDEX IF NOT EXISTS idx_memory_edges_type ON memory_edges(edge_type); + +-- ============================================================================ +-- RL (Reinforcement Learning) Support Tables +-- ============================================================================ +-- Task baselines: per-agent performance statistics for RL credit assignment +CREATE TABLE IF NOT EXISTS task_baselines ( + agent_id TEXT PRIMARY KEY, + count INTEGER DEFAULT 0, + mean_tokens INTEGER DEFAULT 0, + mean_errors REAL DEFAULT 0, + mean_user_corrections REAL DEFAULT 0, + m2_tokens REAL DEFAULT 0, + m2_errors REAL DEFAULT 0, + m2_user_corrections REAL DEFAULT 0, + updated_at DATETIME +); + +-- Task completions: record of completed agent runs for RL analysis +CREATE TABLE IF NOT EXISTS task_completions ( + id BLOB PRIMARY KEY, + agent_id TEXT NOT NULL, + conversation_id BLOB NOT NULL REFERENCES agent_conversations(id) ON DELETE CASCADE, + run_id BLOB NOT NULL REFERENCES agent_runs(id) ON DELETE CASCADE, + description TEXT NOT NULL DEFAULT '', + tokens_used INTEGER DEFAULT 0, + tool_calls INTEGER DEFAULT 0, + errors INTEGER DEFAULT 0, + user_corrections INTEGER DEFAULT 0, + completed BOOLEAN NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); +CREATE INDEX IF NOT EXISTS idx_task_completions_agent_created ON task_completions(agent_id, created_at); +CREATE INDEX IF NOT EXISTS idx_task_completions_run ON task_completions(run_id); + +-- Task retrievals: links memories retrieved during task execution for RL credit assignment +CREATE TABLE IF NOT EXISTS task_retrievals ( + id BLOB PRIMARY KEY, + task_id BLOB NOT NULL REFERENCES task_completions(id) ON DELETE CASCADE, + memory_id BLOB NOT NULL REFERENCES recall_items(id) ON DELETE CASCADE, + similarity REAL NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE(task_id, memory_id) +); +CREATE INDEX IF NOT EXISTS idx_task_retrievals_task ON task_retrievals(task_id); +CREATE INDEX IF NOT EXISTS idx_task_retrievals_memory ON task_retrievals(memory_id); \ No newline at end of file diff --git a/pkg/memory/sqlc/soft_delete.sql.go b/pkg/memory/sqlc/soft_delete.sql.go new file mode 100644 index 000000000..28f9f6f30 --- /dev/null +++ b/pkg/memory/sqlc/soft_delete.sql.go @@ -0,0 +1,310 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: soft_delete.sql + +package sqlc + +import ( + "context" + "time" + + "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + "github.com/ZanzyTHEbar/dragonscale/pkg/memory" +) + +const HardDeleteArchivalChunks = `-- name: HardDeleteArchivalChunks :exec +DELETE FROM archival_chunks +WHERE recall_id = ?1 +` + +type HardDeleteArchivalChunksParams struct { + RecallID ids.UUID `db:"recall_id" json:"recall_id"` +} + +// Permanently deletes archival chunks for a recall item. +// +// DELETE FROM archival_chunks +// WHERE recall_id = ?1 +func (q *Queries) HardDeleteArchivalChunks(ctx context.Context, arg HardDeleteArchivalChunksParams) error { + _, err := q.db.ExecContext(ctx, HardDeleteArchivalChunks, arg.RecallID) + return err +} + +const HardDeleteChunk = `-- name: HardDeleteChunk :exec +DELETE FROM archival_chunks +WHERE id = ?1 +` + +type HardDeleteChunkParams struct { + ID ids.UUID `db:"id" json:"id"` +} + +// Permanently deletes a single archival chunk by ID. +// +// DELETE FROM archival_chunks +// WHERE id = ?1 +func (q *Queries) HardDeleteChunk(ctx context.Context, arg HardDeleteChunkParams) error { + _, err := q.db.ExecContext(ctx, HardDeleteChunk, arg.ID) + return err +} + +const HardDeleteRecallItem = `-- name: HardDeleteRecallItem :exec +DELETE FROM recall_items +WHERE id = ?1 + AND agent_id = ?2 +` + +type HardDeleteRecallItemParams struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Permanently deletes a recall item (after quarantine period). +// +// DELETE FROM recall_items +// WHERE id = ?1 +// AND agent_id = ?2 +func (q *Queries) HardDeleteRecallItem(ctx context.Context, arg HardDeleteRecallItemParams) error { + _, err := q.db.ExecContext(ctx, HardDeleteRecallItem, arg.ID, arg.AgentID) + return err +} + +const ListQuarantinedArchivalChunks = `-- name: ListQuarantinedArchivalChunks :many +SELECT id, + recall_id, + chunk_index, + content, + embedding, + source, + hash, + created_at, + suppressed_at +FROM archival_chunks +WHERE suppressed_at IS NOT NULL + AND suppressed_at < ?1 +ORDER BY suppressed_at ASC +LIMIT ?2 +` + +type ListQuarantinedArchivalChunksParams struct { + BeforeDate *time.Time `db:"before_date" json:"before_date"` + Lim int64 `db:"lim" json:"lim"` +} + +// Returns archival chunks that have been soft-deleted and are +// eligible for permanent deletion. +// +// SELECT id, +// recall_id, +// chunk_index, +// content, +// embedding, +// source, +// hash, +// created_at, +// suppressed_at +// FROM archival_chunks +// WHERE suppressed_at IS NOT NULL +// AND suppressed_at < ?1 +// ORDER BY suppressed_at ASC +// LIMIT ?2 +func (q *Queries) ListQuarantinedArchivalChunks(ctx context.Context, arg ListQuarantinedArchivalChunksParams) ([]ArchivalChunk, error) { + rows, err := q.db.QueryContext(ctx, ListQuarantinedArchivalChunks, arg.BeforeDate, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ArchivalChunk{} + for rows.Next() { + var i ArchivalChunk + if err := rows.Scan( + &i.ID, + &i.RecallID, + &i.ChunkIndex, + &i.Content, + &i.Embedding, + &i.Source, + &i.Hash, + &i.CreatedAt, + &i.SuppressedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const ListQuarantinedRecallItems = `-- name: ListQuarantinedRecallItems :many +SELECT id, + agent_id, + session_key, + role, + sector, + importance, + salience, + decay_rate, + content, + tags, + created_at, + updated_at, + suppressed_at +FROM recall_items +WHERE suppressed_at IS NOT NULL + AND suppressed_at < ?1 +ORDER BY suppressed_at ASC +LIMIT ?2 +` + +type ListQuarantinedRecallItemsParams struct { + BeforeDate *time.Time `db:"before_date" json:"before_date"` + Lim int64 `db:"lim" json:"lim"` +} + +type ListQuarantinedRecallItemsRow struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` + SessionKey string `db:"session_key" json:"session_key"` + Role string `db:"role" json:"role"` + Sector memory.Sector `db:"sector" json:"sector"` + Importance float64 `db:"importance" json:"importance"` + Salience float64 `db:"salience" json:"salience"` + DecayRate float64 `db:"decay_rate" json:"decay_rate"` + Content string `db:"content" json:"content"` + Tags string `db:"tags" json:"tags"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` + SuppressedAt *time.Time `db:"suppressed_at" json:"suppressed_at"` +} + +// Returns recall items that have been soft-deleted (suppressed) and are +// eligible for permanent deletion (older than quarantine period). +// +// SELECT id, +// agent_id, +// session_key, +// role, +// sector, +// importance, +// salience, +// decay_rate, +// content, +// tags, +// created_at, +// updated_at, +// suppressed_at +// FROM recall_items +// WHERE suppressed_at IS NOT NULL +// AND suppressed_at < ?1 +// ORDER BY suppressed_at ASC +// LIMIT ?2 +func (q *Queries) ListQuarantinedRecallItems(ctx context.Context, arg ListQuarantinedRecallItemsParams) ([]ListQuarantinedRecallItemsRow, error) { + rows, err := q.db.QueryContext(ctx, ListQuarantinedRecallItems, arg.BeforeDate, arg.Lim) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListQuarantinedRecallItemsRow{} + for rows.Next() { + var i ListQuarantinedRecallItemsRow + if err := rows.Scan( + &i.ID, + &i.AgentID, + &i.SessionKey, + &i.Role, + &i.Sector, + &i.Importance, + &i.Salience, + &i.DecayRate, + &i.Content, + &i.Tags, + &i.CreatedAt, + &i.UpdatedAt, + &i.SuppressedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const RestoreRecallItem = `-- name: RestoreRecallItem :exec +UPDATE recall_items +SET suppressed_at = NULL +WHERE id = ?1 + AND agent_id = ?2 +` + +type RestoreRecallItemParams struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Restores a soft-deleted recall item by clearing suppressed_at. +// +// UPDATE recall_items +// SET suppressed_at = NULL +// WHERE id = ?1 +// AND agent_id = ?2 +func (q *Queries) RestoreRecallItem(ctx context.Context, arg RestoreRecallItemParams) error { + _, err := q.db.ExecContext(ctx, RestoreRecallItem, arg.ID, arg.AgentID) + return err +} + +const SoftDeleteArchivalChunks = `-- name: SoftDeleteArchivalChunks :exec +UPDATE archival_chunks +SET suppressed_at = datetime('now') +WHERE recall_id = ?1 +` + +type SoftDeleteArchivalChunksParams struct { + RecallID ids.UUID `db:"recall_id" json:"recall_id"` +} + +// Sets suppressed_at for all chunks belonging to a recall item. +// +// UPDATE archival_chunks +// SET suppressed_at = datetime('now') +// WHERE recall_id = ?1 +func (q *Queries) SoftDeleteArchivalChunks(ctx context.Context, arg SoftDeleteArchivalChunksParams) error { + _, err := q.db.ExecContext(ctx, SoftDeleteArchivalChunks, arg.RecallID) + return err +} + +const SoftDeleteRecallItem = `-- name: SoftDeleteRecallItem :exec +UPDATE recall_items +SET suppressed_at = datetime('now') +WHERE id = ?1 + AND agent_id = ?2 +` + +type SoftDeleteRecallItemParams struct { + ID ids.UUID `db:"id" json:"id"` + AgentID string `db:"agent_id" json:"agent_id"` +} + +// Soft delete queries (T2.4: quarantine before permanent deletion) +// Sets suppressed_at for a recall item instead of permanently deleting it. +// +// UPDATE recall_items +// SET suppressed_at = datetime('now') +// WHERE id = ?1 +// AND agent_id = ?2 +func (q *Queries) SoftDeleteRecallItem(ctx context.Context, arg SoftDeleteRecallItemParams) error { + _, err := q.db.ExecContext(ctx, SoftDeleteRecallItem, arg.ID, arg.AgentID) + return err +} diff --git a/pkg/memory/sqlc/sqlc.yaml b/pkg/memory/sqlc/sqlc.yaml index c92ffdbc9..fd6a7172d 100644 --- a/pkg/memory/sqlc/sqlc.yaml +++ b/pkg/memory/sqlc/sqlc.yaml @@ -208,6 +208,46 @@ sql: go_type: import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" type: "UUID" + # Immutable messages + - column: "immutable_messages.id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + # Memory edges — from_id/to_id reference recall_items + - column: "memory_edges.from_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + - column: "memory_edges.to_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + # Task completions + - column: "task_completions.id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + - column: "task_completions.conversation_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + - column: "task_completions.run_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + # Task retrievals + - column: "task_retrievals.id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + - column: "task_retrievals.task_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" + - column: "task_retrievals.memory_id" + go_type: + import: "github.com/ZanzyTHEbar/dragonscale/pkg/ids" + type: "UUID" # DAG - column: "dag_snapshots.id" go_type: