opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(memory): update delegate, sqlc queries (agent_audit_log, recall, rl)

Browse source
This commit is contained in:
ZanzyTHEbar 2026-03-05 21:16:49 +00:00
parent b9208c0bf8
commit ab4655901a
9 changed files with 563 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

174
pkg/memory/delegate/rl_store_test.go
View file

@ -180,17 +180,10 @@ func TestSQLiteDelegate_UpdateMemoryWeight(t *testing.T) {
// Insert a recall item first
memoryID := insertTestRecallItem(ctx, t, d, agentID)
// Update the memory weight via direct query
// Update the memory weight via delegate API
newWeight := 2.5
credit := 3.0
paramsRLWeight := newWeight
paramsRLCredit := credit
err := d.Queries().UpdateMemoryWeight(ctx, memsqlc.UpdateMemoryWeightParams{
RlWeight: &paramsRLWeight,
RlCredit: &paramsRLCredit,
ID: memoryID,
AgentID: agentID,
})
err := d.UpdateMemoryWeight(ctx, memoryID, newWeight, credit)
if err != nil {
t.Fatalf("UpdateMemoryWeight: %v", err)
}
@ -209,21 +202,12 @@ func TestSQLiteDelegate_UpdateMemoryWeight_NonExistent(t *testing.T) {
t.Parallel()
d := setupRLTest(t)
ctx := t.Context()
agentID := "test-agent"
// Try to update weight for non-existent memory via direct query
// Try to update weight for non-existent memory.
nonExistentID := ids.New()
paramsWeight := 2.0
paramsCredit := 1.0
err := d.Queries().UpdateMemoryWeight(ctx, memsqlc.UpdateMemoryWeightParams{
RlWeight: &paramsWeight,
RlCredit: &paramsCredit,
ID: nonExistentID,
AgentID: agentID,
})
// Query succeeds but doesn't update anything (no error for non-existent)
if err != nil {
t.Errorf("UpdateMemoryWeight should not error for non-existent: %v", err)
err := d.UpdateMemoryWeight(ctx, nonExistentID, 2.0, 1.0)
if err == nil {
t.Fatal("expected error for non-existent memory update")
}
}
@ -236,13 +220,8 @@ func TestSQLiteDelegate_UpdateMemorySelfReport(t *testing.T) {
// Insert a recall item first
memoryID := insertTestRecallItem(ctx, t, d, agentID)
// Update self-report score via direct query
score := int64(2)
err := d.Queries().UpdateMemorySelfReportScore(ctx, memsqlc.UpdateMemorySelfReportScoreParams{
SelfReportScore: &score,
ID: memoryID,
AgentID: agentID,
})
// Update self-report score via delegate API.
err := d.UpdateMemorySelfReport(ctx, memoryID, 2)
if err != nil {
t.Fatalf("UpdateMemorySelfReportScore: %v", err)
}
@ -261,19 +240,12 @@ func TestSQLiteDelegate_UpdateMemorySelfReport_NonExistent(t *testing.T) {
t.Parallel()
d := setupRLTest(t)
ctx := t.Context()
agentID := "test-agent"
// Try to update self-report for non-existent memory via direct query
// Try to update self-report for non-existent memory.
nonExistentID := ids.New()
score := int64(3)
err := d.Queries().UpdateMemorySelfReportScore(ctx, memsqlc.UpdateMemorySelfReportScoreParams{
SelfReportScore: &score,
ID: nonExistentID,
AgentID: agentID,
})
// Query succeeds but doesn't update anything (no error for non-existent)
if err != nil {
t.Errorf("UpdateMemorySelfReportScore should not error for non-existent: %v", err)
err := d.UpdateMemorySelfReport(ctx, nonExistentID, 3)
if err == nil {
t.Fatal("expected error for non-existent memory self-report update")
}
}
@ -547,6 +519,7 @@ func TestSQLiteDelegate_GetRecentAuditEntries(t *testing.T) {
t.Parallel()
d := setupRLTest(t)
ctx := t.Context()
since := time.Now().Add(-1 * time.Minute)
// Insert some audit entries
entries := []*memory.AuditEntry{
@ -561,8 +534,8 @@ func TestSQLiteDelegate_GetRecentAuditEntries(t *testing.T) {
},
{
ID: ids.New(),
AgentID: "audit-agent",
SessionKey: "session-1",
AgentID: "audit-agent-2",
SessionKey: "session-2",
Action: "write_file",
Target: "/path/to/output",
Input: `{"path": "/output"}`,
@ -576,15 +549,27 @@ func TestSQLiteDelegate_GetRecentAuditEntries(t *testing.T) {
}
}
// Get recent audit entries (all of them, since time is in the past)
auditEntries, err := d.GetRecentAuditEntries(ctx, time.Time{})
// Get recent audit entries across all agents.
auditEntries, err := d.GetRecentAuditEntries(ctx, since)
if err != nil {
t.Fatalf("GetRecentAuditEntries: %v", err)
}
if len(auditEntries) != 2 {
t.Fatalf("expected 2 recent audit entries, got %d", len(auditEntries))
}
// The implementation uses ListAuditEntries with empty agent_id which may filter results
// Just verify the query executes without error
t.Logf("Got %d audit entries", len(auditEntries))
foundAgents := map[string]bool{}
foundTools := map[string]bool{}
for _, entry := range auditEntries {
foundAgents[entry.AgentID] = true
foundTools[entry.ToolName] = true
}
if !foundAgents["audit-agent"] || !foundAgents["audit-agent-2"] {
t.Fatalf("expected entries from both agents, got %+v", foundAgents)
}
if !foundTools["read_file"] || !foundTools["write_file"] {
t.Fatalf("expected tool names mapped from action, got %+v", foundTools)
}
}
func TestSQLiteDelegate_GetHighTokenSessions(t *testing.T) {
@ -592,13 +577,79 @@ func TestSQLiteDelegate_GetHighTokenSessions(t *testing.T) {
d := setupRLTest(t)
ctx := t.Context()
// This is a placeholder implementation that returns empty list
sessions, err := d.GetHighTokenSessions(ctx, 1000)
// Recreate task_completions with relaxed constraints for focused aggregation testing.
_, _ = d.db.ExecContext(ctx, `DROP TABLE IF EXISTS task_completions`)
_, err := d.db.ExecContext(ctx, `
CREATE TABLE task_completions (
id BLOB PRIMARY KEY,
agent_id TEXT NOT NULL,
conversation_id BLOB NOT NULL,
run_id BLOB NOT NULL,
description TEXT,
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'))
) WITHOUT ROWID
`)
if err != nil {
t.Fatalf("create task_completions table: %v", err)
}
session1 := ids.New()
session2 := ids.New()
session3 := ids.New()
runID := ids.New()
_, err = d.db.ExecContext(ctx, `
INSERT INTO task_completions (id, agent_id, conversation_id, run_id, description, tokens_used, completed)
VALUES (?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?)
`,
ids.New(), "agent-a", session1, runID, "task-1", 800, true,
ids.New(), "agent-a", session1, runID, "task-2", 900, true,
ids.New(), "agent-b", session2, runID, "task-3", 400, true,
ids.New(), "agent-c", session3, runID, "task-4", 300, true,
ids.New(), "agent-c", session3, runID, "task-5", 450, true,
)
if err != nil {
t.Fatalf("insert task completions: %v", err)
}
sessions, err := d.GetHighTokenSessions(ctx, 700)
if err != nil {
t.Fatalf("GetHighTokenSessions: %v", err)
}
if len(sessions) != 0 {
t.Errorf("expected 0 sessions (placeholder), got %d", len(sessions))
if len(sessions) != 2 {
t.Fatalf("expected 2 high-token sessions, got %d", len(sessions))
}
gotByAgent := map[string]SessionSummary{}
for _, s := range sessions {
gotByAgent[s.AgentID] = s
}
a, ok := gotByAgent["agent-a"]
if !ok {
t.Fatalf("expected aggregated session for agent-a, got %+v", gotByAgent)
}
if a.SessionID != session1.String() {
t.Fatalf("expected session %s for agent-a, got %s", session1.String(), a.SessionID)
}
if a.TotalTokens < 1700 {
t.Fatalf("expected aggregated tokens >= 1700 for agent-a, got %d", a.TotalTokens)
}
c, ok := gotByAgent["agent-c"]
if !ok {
t.Fatalf("expected aggregated session for agent-c, got %+v", gotByAgent)
}
if c.SessionID != session3.String() {
t.Fatalf("expected session %s for agent-c, got %s", session3.String(), c.SessionID)
}
if c.TotalTokens < 750 {
t.Fatalf("expected aggregated tokens >= 750 for agent-c, got %d", c.TotalTokens)
}
}
@ -654,19 +705,11 @@ func TestSQLiteDelegate_RLStore_Integration(t *testing.T) {
memoryIDs[i] = insertTestRecallItem(ctx, t, d, agentID)
}
// Update weights for each memory via direct query
// Note: GetRecallItem doesn't return RL fields, so we just verify no errors
// Update weights for each memory via delegate API.
weights := []float64{1.5, 2.0, 2.5}
credits := []float64{1.0, 2.0, 3.0}
for i, memoryID := range memoryIDs {
paramsRLWeight := weights[i]
paramsRLCredit := credits[i]
err := d.Queries().UpdateMemoryWeight(ctx, memsqlc.UpdateMemoryWeightParams{
RlWeight: &paramsRLWeight,
RlCredit: &paramsRLCredit,
ID: memoryID,
AgentID: agentID,
})
err := d.UpdateMemoryWeight(ctx, memoryID, weights[i], credits[i])
if err != nil {
t.Fatalf("UpdateMemoryWeight %d: %v", i, err)
}
@ -687,15 +730,10 @@ func TestSQLiteDelegate_RLStore_Integration(t *testing.T) {
t.Run("SelfReportUpdates", func(t *testing.T) {
memoryID := insertTestRecallItem(ctx, t, d, agentID)
// Update self-report scores via direct query
// Note: GetRecallItem doesn't return self_report_score, so we just verify no errors
// Update self-report scores via delegate API.
scores := []int64{0, 1, 2, 3}
for _, score := range scores {
err := d.Queries().UpdateMemorySelfReportScore(ctx, memsqlc.UpdateMemorySelfReportScoreParams{
SelfReportScore: &score,
ID: memoryID,
AgentID: agentID,
})
err := d.UpdateMemorySelfReport(ctx, memoryID, int(score))
if err != nil {
t.Fatalf("UpdateMemorySelfReportScore %d: %v", score, err)
}

110
pkg/memory/delegate/sqlite.go
View file

@ -5,6 +5,7 @@ import (
"context"
"database/sql"
"fmt"
"strings"
"time"
"github.com/ZanzyTHEbar/dragonscale/pkg/ids"
@ -1079,19 +1080,20 @@ func (d *LibSQLDelegate) UpdateTaskBaseline(ctx context.Context, agentID string,
func (d *LibSQLDelegate) UpdateMemoryWeight(ctx context.Context, memoryID ids.UUID, weight, credit float64) error {
rlWeight := weight
rlCredit := credit
// Get the agent_id from the memory item first
item, err := d.GetRecallItem(ctx, "", memoryID)
row, err := d.queries.GetRecallItemByID(ctx, memsqlc.GetRecallItemByIDParams{
ID: memoryID,
})
if err == sql.ErrNoRows {
return fmt.Errorf("memory item not found: %s", memoryID)
}
if err != nil {
return err
}
if item == nil {
return fmt.Errorf("memory item not found: %s", memoryID)
}
return d.queries.UpdateMemoryWeight(ctx, memsqlc.UpdateMemoryWeightParams{
RlWeight: &rlWeight,
RlCredit: &rlCredit,
ID: memoryID,
AgentID: item.AgentID,
AgentID: row.AgentID,
})
}
@ -1099,18 +1101,19 @@ func (d *LibSQLDelegate) UpdateMemoryWeight(ctx context.Context, memoryID ids.UU
// Implements cortex.RLStore interface.
func (d *LibSQLDelegate) UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error {
selfReportScore := int64(score)
// Get the agent_id from the memory item first
item, err := d.GetRecallItem(ctx, "", memoryID)
row, err := d.queries.GetRecallItemByID(ctx, memsqlc.GetRecallItemByIDParams{
ID: memoryID,
})
if err == sql.ErrNoRows {
return fmt.Errorf("memory item not found: %s", memoryID)
}
if err != nil {
return err
}
if item == nil {
return fmt.Errorf("memory item not found: %s", memoryID)
}
return d.queries.UpdateMemorySelfReportScore(ctx, memsqlc.UpdateMemorySelfReportScoreParams{
SelfReportScore: &selfReportScore,
ID: memoryID,
AgentID: item.AgentID,
AgentID: row.AgentID,
})
}
@ -1207,33 +1210,67 @@ func (d *LibSQLDelegate) ListActiveAgents(ctx context.Context, since time.Time)
// GetRecentAuditEntries returns audit entries since the given time.
// Implements cortex.AuditAnalysisStore interface.
func (d *LibSQLDelegate) GetRecentAuditEntries(ctx context.Context, since time.Time) ([]AuditEntry, error) {
// Get all audit entries and filter by time
rows, err := d.queries.ListAuditEntries(ctx, memsqlc.ListAuditEntriesParams{
AgentID: "", // Get all agents
Lim: 10000,
})
if err != nil {
return nil, err
cutoff := since.UTC()
if cutoff.IsZero() {
cutoff = time.Unix(0, 0).UTC()
}
const pageSize int64 = 1000
offset := int64(0)
entries := make([]AuditEntry, 0, pageSize)
for {
rows, err := d.queries.ListAuditEntriesGlobalSincePaged(ctx, memsqlc.ListAuditEntriesGlobalSincePagedParams{
Since: cutoff,
Lim: pageSize,
Off: offset,
})
if err != nil {
return nil, fmt.Errorf("list audit entries since: %w", err)
}
if len(rows) == 0 {
break
}
var entries []AuditEntry
for _, row := range rows {
if row.CreatedAt.After(since) {
lowerAction := strings.ToLower(strings.TrimSpace(row.Action))
toolName := strings.TrimSpace(row.Action)
if strings.HasPrefix(lowerAction, "tool_") && strings.TrimSpace(row.Target) != "" {
toolName = strings.TrimSpace(row.Target)
}
if toolName == "" {
toolName = strings.TrimSpace(row.Target)
}
success := true
if lowerAction == "tool_error" || strings.Contains(lowerAction, "error") || strings.Contains(lowerAction, "fail") {
success = false
}
entry := AuditEntry{
ID: row.ID.String(),
Timestamp: row.CreatedAt,
ToolName: row.Action, // Using action as tool name proxy
ToolName: toolName,
ToolInput: "",
Success: true, // Default to success
Success: success,
SessionID: row.SessionKey,
AgentID: row.AgentID,
}
if row.Input != nil {
entry.ToolInput = *row.Input
}
if !success && row.Output != nil {
entry.ErrorMsg = *row.Output
}
entries = append(entries, entry)
}
if len(rows) < int(pageSize) {
break
}
offset += int64(len(rows))
}
return entries, nil
}
@ -1258,10 +1295,31 @@ func (d *LibSQLDelegate) StoreDetectedPattern(ctx context.Context, pattern Detec
// GetHighTokenSessions returns sessions with token usage above threshold.
// Implements cortex.AuditAnalysisStore interface.
// Note: This is a placeholder - actual token tracking needs to be implemented.
func (d *LibSQLDelegate) GetHighTokenSessions(ctx context.Context, minTokens int64) ([]SessionSummary, error) {
// TODO: Implement token-based session filtering when token tracking is available
return []SessionSummary{}, nil
min := minTokens
rows, err := d.queries.GetHighTokenSessions(ctx, memsqlc.GetHighTokenSessionsParams{
MinTokens: &min,
Lim: 100,
})
if err != nil {
return nil, fmt.Errorf("get high token sessions: %w", err)
}
summaries := make([]SessionSummary, 0, len(rows))
for _, row := range rows {
totalTokens := int64(0)
if row.TotalTokens != nil {
totalTokens = int64(*row.TotalTokens)
}
summaries = append(summaries, SessionSummary{
SessionID: row.SessionID.String(),
AgentID: row.AgentID,
TotalTokens: totalTokens,
ToolCounts: map[string]int{},
})
}
return summaries, nil
}
// --- Batch Operations for Cortex Tasks (via sqlc) ---

160
pkg/memory/sqlc/agent_audit_log.sql.go
View file

@ -80,9 +80,17 @@ VALUES (
?6,
?7,
?8,
datetime('now')
strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
)
RETURNING id, agent_id, session_key, action, target, input, output, duration_ms, created_at
RETURNING id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at
`
type InsertAuditEntryParams struct {
@ -118,9 +126,17 @@ type InsertAuditEntryParams struct {
// ?6,
// ?7,
// ?8,
// datetime('now')
// strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
// )
// RETURNING id, agent_id, session_key, action, target, input, output, duration_ms, created_at
// RETURNING id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
func (q *Queries) InsertAuditEntry(ctx context.Context, arg InsertAuditEntryParams) (AgentAuditLog, error) {
row := q.db.QueryRowContext(ctx, InsertAuditEntry,
arg.ID,
@ -360,6 +376,142 @@ func (q *Queries) ListAuditEntriesBySession(ctx context.Context, arg ListAuditEn
return items, nil
}
const ListAuditEntriesGlobal = `-- name: ListAuditEntriesGlobal :many
SELECT id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at
FROM agent_audit_log
ORDER BY created_at DESC
LIMIT ?1
`
type ListAuditEntriesGlobalParams struct {
Lim int64 `db:"lim" json:"lim"`
}
// ListAuditEntriesGlobal
//
// SELECT id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
// FROM agent_audit_log
// ORDER BY created_at DESC
// LIMIT ?1
func (q *Queries) ListAuditEntriesGlobal(ctx context.Context, arg ListAuditEntriesGlobalParams) ([]AgentAuditLog, error) {
rows, err := q.db.QueryContext(ctx, ListAuditEntriesGlobal, arg.Lim)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentAuditLog{}
for rows.Next() {
var i AgentAuditLog
if err := rows.Scan(
&i.ID,
&i.AgentID,
&i.SessionKey,
&i.Action,
&i.Target,
&i.Input,
&i.Output,
&i.DurationMs,
&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 ListAuditEntriesGlobalSincePaged = `-- name: ListAuditEntriesGlobalSincePaged :many
SELECT id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at
FROM agent_audit_log
WHERE julianday(created_at) > julianday(?1)
ORDER BY created_at ASC, id ASC
LIMIT ?3 OFFSET ?2
`
type ListAuditEntriesGlobalSincePagedParams struct {
Since interface{} `db:"since" json:"since"`
Off int64 `db:"off" json:"off"`
Lim int64 `db:"lim" json:"lim"`
}
// ListAuditEntriesGlobalSincePaged
//
// SELECT id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
// FROM agent_audit_log
// WHERE julianday(created_at) > julianday(?1)
// ORDER BY created_at ASC, id ASC
// LIMIT ?3 OFFSET ?2
func (q *Queries) ListAuditEntriesGlobalSincePaged(ctx context.Context, arg ListAuditEntriesGlobalSincePagedParams) ([]AgentAuditLog, error) {
rows, err := q.db.QueryContext(ctx, ListAuditEntriesGlobalSincePaged, arg.Since, arg.Off, arg.Lim)
if err != nil {
return nil, err
}
defer rows.Close()
items := []AgentAuditLog{}
for rows.Next() {
var i AgentAuditLog
if err := rows.Scan(
&i.ID,
&i.AgentID,
&i.SessionKey,
&i.Action,
&i.Target,
&i.Input,
&i.Output,
&i.DurationMs,
&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 PruneOldAuditEntries = `-- name: PruneOldAuditEntries :exec
DELETE FROM agent_audit_log
WHERE agent_id = ?1

69
pkg/memory/sqlc/querier.go
View file

@ -509,13 +509,12 @@ type Querier interface {
// SELECT
// conversation_id as session_id,
// agent_id,
// SUM(tokens_used) as total_tokens,
// SUM(COALESCE(tokens_used, 0)) as total_tokens,
// COUNT(*) as task_count
// FROM task_completions
// WHERE tokens_used > ?1
// AND created_at > datetime('now', '-24 hours')
// WHERE created_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-24 hours')
// GROUP BY conversation_id, agent_id
// HAVING total_tokens > ?1
// HAVING SUM(COALESCE(tokens_used, 0)) > ?1
// ORDER BY total_tokens DESC
// LIMIT ?2
GetHighTokenSessions(ctx context.Context, arg GetHighTokenSessionsParams) ([]GetHighTokenSessionsRow, error)
@ -661,6 +660,25 @@ type Querier interface {
// AND suppressed_at IS NULL
// LIMIT 1
GetRecallItem(ctx context.Context, arg GetRecallItemParams) (GetRecallItemRow, error)
//GetRecallItemByID
//
// SELECT id,
// agent_id,
// session_key,
// role,
// sector,
// importance,
// salience,
// decay_rate,
// content,
// tags,
// created_at,
// updated_at
// FROM recall_items
// WHERE id = ?1
// AND suppressed_at IS NULL
// LIMIT 1
GetRecallItemByID(ctx context.Context, arg GetRecallItemByIDParams) (GetRecallItemByIDRow, error)
//GetRecallItemsByIDs
//
// SELECT id,
@ -788,9 +806,17 @@ type Querier interface {
// ?6,
// ?7,
// ?8,
// datetime('now')
// strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
// )
// RETURNING id, agent_id, session_key, action, target, input, output, duration_ms, created_at
// RETURNING id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
InsertAuditEntry(ctx context.Context, arg InsertAuditEntryParams) (AgentAuditLog, error)
//InsertDAGEdge
//
@ -1355,6 +1381,37 @@ type Querier interface {
// ORDER BY created_at DESC
// LIMIT ?3
ListAuditEntriesBySession(ctx context.Context, arg ListAuditEntriesBySessionParams) ([]AgentAuditLog, error)
//ListAuditEntriesGlobal
//
// SELECT id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
// FROM agent_audit_log
// ORDER BY created_at DESC
// LIMIT ?1
ListAuditEntriesGlobal(ctx context.Context, arg ListAuditEntriesGlobalParams) ([]AgentAuditLog, error)
//ListAuditEntriesGlobalSincePaged
//
// SELECT id,
// agent_id,
// session_key,
// action,
// target,
// input,
// output,
// duration_ms,
// created_at
// FROM agent_audit_log
// WHERE julianday(created_at) > julianday(?1)
// ORDER BY created_at ASC, id ASC
// LIMIT ?3 OFFSET ?2
ListAuditEntriesGlobalSincePaged(ctx context.Context, arg ListAuditEntriesGlobalSincePagedParams) ([]AgentAuditLog, error)
//ListDAGEdgesBySnapshotID
//
// SELECT id,

40
pkg/memory/sqlc/queries/agent_audit_log.sql
View file

@ -20,9 +20,17 @@ VALUES (
sqlc.arg(input),
sqlc.arg(output),
sqlc.arg(duration_ms),
datetime('now')
strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
)
RETURNING id, agent_id, session_key, action, target, input, output, duration_ms, created_at;
RETURNING id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at;
-- name: ListAuditEntries :many
SELECT id,
agent_id,
@ -37,6 +45,34 @@ FROM agent_audit_log
WHERE agent_id = sqlc.arg(agent_id)
ORDER BY created_at DESC
LIMIT sqlc.arg(lim);
-- name: ListAuditEntriesGlobal :many
SELECT id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at
FROM agent_audit_log
ORDER BY created_at DESC
LIMIT sqlc.arg(lim);
-- name: ListAuditEntriesGlobalSincePaged :many
SELECT id,
agent_id,
session_key,
action,
target,
input,
output,
duration_ms,
created_at
FROM agent_audit_log
WHERE julianday(created_at) > julianday(sqlc.arg(since))
ORDER BY created_at ASC,
id ASC
LIMIT sqlc.arg(lim) OFFSET sqlc.arg(off);
-- name: ListAuditEntriesByAction :many
SELECT id,
agent_id,

17
pkg/memory/sqlc/queries/recall.sql
View file

@ -61,6 +61,23 @@ WHERE id = sqlc.arg(id)
AND agent_id = sqlc.arg(agent_id)
AND suppressed_at IS NULL
LIMIT 1;
-- name: GetRecallItemByID :one
SELECT id,
agent_id,
session_key,
role,
sector,
importance,
salience,
decay_rate,
content,
tags,
created_at,
updated_at
FROM recall_items
WHERE id = sqlc.arg(id)
AND suppressed_at IS NULL
LIMIT 1;
-- name: UpdateRecallItem :exec
UPDATE recall_items
SET role = sqlc.arg(role),

40
pkg/memory/sqlc/queries/rl.sql
View file

@ -1,6 +1,5 @@
-- 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,
@ -15,7 +14,6 @@ SELECT agent_id,
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 (
@ -39,8 +37,7 @@ VALUES (
sqlc.arg(m2_errors),
sqlc.arg(m2_user_corrections),
datetime('now')
)
ON CONFLICT (agent_id) DO
) ON CONFLICT (agent_id) DO
UPDATE
SET count = excluded.count,
mean_tokens = excluded.mean_tokens,
@ -50,7 +47,6 @@ SET count = excluded.count,
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
@ -59,7 +55,6 @@ SET rl_weight = sqlc.arg(rl_weight),
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
@ -67,7 +62,6 @@ 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
@ -75,7 +69,6 @@ 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,
@ -99,7 +92,6 @@ 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,
@ -127,7 +119,6 @@ WHERE agent_id = sqlc.arg(agent_id)
)
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 (
@ -165,7 +156,6 @@ RETURNING id,
user_corrections,
completed,
created_at;
-- name: GetCompletedTasks :many
-- Get tasks completed since the given time for RL processing
SELECT id,
@ -184,14 +174,17 @@ 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;
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,
@ -200,25 +193,22 @@ SELECT tr.memory_id,
FROM task_retrievals tr
JOIN recall_items ri ON tr.memory_id = ri.id
WHERE tr.task_id = sqlc.arg(task_id);
-- name: ListActiveAgents :many
-- Get all unique agent IDs that have completed tasks (for multi-agent processing)
SELECT DISTINCT agent_id
FROM task_completions
WHERE created_at > sqlc.arg(since)
ORDER BY agent_id;
-- name: GetHighTokenSessions :many
-- Get sessions with high token usage grouped by conversation/agent
SELECT
conversation_id as session_id,
SELECT conversation_id as session_id,
agent_id,
SUM(tokens_used) as total_tokens,
SUM(COALESCE(tokens_used, 0)) as total_tokens,
COUNT(*) as task_count
FROM task_completions
WHERE tokens_used > sqlc.arg(min_tokens)
AND created_at > datetime('now', '-24 hours')
GROUP BY conversation_id, agent_id
HAVING total_tokens > sqlc.arg(min_tokens)
WHERE created_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-24 hours')
GROUP BY conversation_id,
agent_id
HAVING SUM(COALESCE(tokens_used, 0)) > sqlc.arg(min_tokens)
ORDER BY total_tokens DESC
LIMIT sqlc.arg(lim);

76
pkg/memory/sqlc/recall.sql.go
View file

@ -174,6 +174,82 @@ func (q *Queries) GetRecallItem(ctx context.Context, arg GetRecallItemParams) (G
return i, err
}
const GetRecallItemByID = `-- name: GetRecallItemByID :one
SELECT id,
agent_id,
session_key,
role,
sector,
importance,
salience,
decay_rate,
content,
tags,
created_at,
updated_at
FROM recall_items
WHERE id = ?1
AND suppressed_at IS NULL
LIMIT 1
`
type GetRecallItemByIDParams struct {
ID ids.UUID `db:"id" json:"id"`
}
type GetRecallItemByIDRow 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"`
}
// GetRecallItemByID
//
// SELECT id,
// agent_id,
// session_key,
// role,
// sector,
// importance,
// salience,
// decay_rate,
// content,
// tags,
// created_at,
// updated_at
// FROM recall_items
// WHERE id = ?1
// AND suppressed_at IS NULL
// LIMIT 1
func (q *Queries) GetRecallItemByID(ctx context.Context, arg GetRecallItemByIDParams) (GetRecallItemByIDRow, error) {
row := q.db.QueryRowContext(ctx, GetRecallItemByID, arg.ID)
var i GetRecallItemByIDRow
err := row.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,
)
return i, err
}
const GetRecallItemsByIDs = `-- name: GetRecallItemsByIDs :many
SELECT id,
agent_id,

14
pkg/memory/sqlc/rl.sql.go
View file

@ -94,13 +94,12 @@ const GetHighTokenSessions = `-- name: GetHighTokenSessions :many
SELECT
conversation_id as session_id,
agent_id,
SUM(tokens_used) as total_tokens,
SUM(COALESCE(tokens_used, 0)) as total_tokens,
COUNT(*) as task_count
FROM task_completions
WHERE tokens_used > ?1
AND created_at > datetime('now', '-24 hours')
WHERE created_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-24 hours')
GROUP BY conversation_id, agent_id
HAVING total_tokens > ?1
HAVING SUM(COALESCE(tokens_used, 0)) > ?1
ORDER BY total_tokens DESC
LIMIT ?2
`
@ -122,13 +121,12 @@ type GetHighTokenSessionsRow struct {
// SELECT
// conversation_id as session_id,
// agent_id,
// SUM(tokens_used) as total_tokens,
// SUM(COALESCE(tokens_used, 0)) as total_tokens,
// COUNT(*) as task_count
// FROM task_completions
// WHERE tokens_used > ?1
// AND created_at > datetime('now', '-24 hours')
// WHERE created_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-24 hours')
// GROUP BY conversation_id, agent_id
// HAVING total_tokens > ?1
// HAVING SUM(COALESCE(tokens_used, 0)) > ?1
// ORDER BY total_tokens DESC
// LIMIT ?2
func (q *Queries) GetHighTokenSessions(ctx context.Context, arg GetHighTokenSessionsParams) ([]GetHighTokenSessionsRow, error) {