diff --git a/pkg/memory/delegate/rl_store_test.go b/pkg/memory/delegate/rl_store_test.go index f37ba145d..b4c9ece3e 100644 --- a/pkg/memory/delegate/rl_store_test.go +++ b/pkg/memory/delegate/rl_store_test.go @@ -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: ¶msRLWeight, - RlCredit: ¶msRLCredit, - 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: ¶msWeight, - RlCredit: ¶msCredit, - 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: ¶msRLWeight, - RlCredit: ¶msRLCredit, - 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) } diff --git a/pkg/memory/delegate/sqlite.go b/pkg/memory/delegate/sqlite.go index c09dad96f..778149cad 100644 --- a/pkg/memory/delegate/sqlite.go +++ b/pkg/memory/delegate/sqlite.go @@ -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() } - var entries []AuditEntry - for _, row := range rows { - if row.CreatedAt.After(since) { + 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 + } + + for _, row := range rows { + 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) --- diff --git a/pkg/memory/sqlc/agent_audit_log.sql.go b/pkg/memory/sqlc/agent_audit_log.sql.go index cecdb46e1..5ff7ee61a 100644 --- a/pkg/memory/sqlc/agent_audit_log.sql.go +++ b/pkg/memory/sqlc/agent_audit_log.sql.go @@ -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 diff --git a/pkg/memory/sqlc/querier.go b/pkg/memory/sqlc/querier.go index 3dc226b88..8236635b5 100644 --- a/pkg/memory/sqlc/querier.go +++ b/pkg/memory/sqlc/querier.go @@ -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, diff --git a/pkg/memory/sqlc/queries/agent_audit_log.sql b/pkg/memory/sqlc/queries/agent_audit_log.sql index d636b42b8..c75e30149 100644 --- a/pkg/memory/sqlc/queries/agent_audit_log.sql +++ b/pkg/memory/sqlc/queries/agent_audit_log.sql @@ -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, diff --git a/pkg/memory/sqlc/queries/recall.sql b/pkg/memory/sqlc/queries/recall.sql index c59fdf6be..925c1f546 100644 --- a/pkg/memory/sqlc/queries/recall.sql +++ b/pkg/memory/sqlc/queries/recall.sql @@ -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), diff --git a/pkg/memory/sqlc/queries/rl.sql b/pkg/memory/sqlc/queries/rl.sql index a97d60ae7..e34f647c6 100644 --- a/pkg/memory/sqlc/queries/rl.sql +++ b/pkg/memory/sqlc/queries/rl.sql @@ -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); +LIMIT sqlc.arg(lim); \ No newline at end of file diff --git a/pkg/memory/sqlc/recall.sql.go b/pkg/memory/sqlc/recall.sql.go index 56d272dea..67b98bfca 100644 --- a/pkg/memory/sqlc/recall.sql.go +++ b/pkg/memory/sqlc/recall.sql.go @@ -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, diff --git a/pkg/memory/sqlc/rl.sql.go b/pkg/memory/sqlc/rl.sql.go index 4161a9596..35a13e438 100644 --- a/pkg/memory/sqlc/rl.sql.go +++ b/pkg/memory/sqlc/rl.sql.go @@ -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) {