P4: contrib/store — persona SQLite store (JSON-blob round-trip)
db.Personas() satisfies persona.Storage over SQLite. Each Agent is stored as a JSON blob with extracted indexed columns (owner_id, name, webhook_secret, chatbot_channel_filter, schedule, next_run_at) — so the WHOLE struct round-trips (no domain<->GORM<->DB field-loss footgun) while the lookups stay indexable. Test proves the round-trip preserves nested + map fields (SkillPalette, StateReactEmoji), the owner/name + webhook + chatbot-filter queries, the scheduled-due query, and MarkAgentScheduledRun clearing the due window. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -76,8 +76,9 @@ BATTERIES (opt-in siblings, each nil-safe + a default):
|
|||||||
BudgetStorage iface + Memory default
|
BudgetStorage iface + Memory default
|
||||||
|
|
||||||
contrib/store/ SECOND module (+ modernc.org/sqlite): [P4 ~]
|
contrib/store/ SECOND module (+ modernc.org/sqlite): [P4 ~]
|
||||||
pure-Go SQLite impls of the *Store seams. budget ✓
|
pure-Go SQLite impls of the *Store seams. budget +
|
||||||
(conformance-tested); persona/skill/audit pending.
|
persona ✓ (JSON-blob+indexed cols, round-trip tested);
|
||||||
|
skill/audit pending.
|
||||||
CI proves the driver lands HERE, not in the core go.sum.
|
CI proves the driver lands HERE, not in the core go.sum.
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ require (
|
|||||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||||
golang.org/x/crypto v0.53.0 // indirect
|
golang.org/x/crypto v0.53.0 // indirect
|
||||||
golang.org/x/sys v0.46.0 // indirect
|
golang.org/x/sys v0.46.0 // indirect
|
||||||
|
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||||
modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
|
modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
|
||||||
modernc.org/libc v1.55.3 // indirect
|
modernc.org/libc v1.55.3 // indirect
|
||||||
modernc.org/mathutil v1.6.0 // indirect
|
modernc.org/mathutil v1.6.0 // indirect
|
||||||
|
|||||||
@@ -25,6 +25,9 @@ golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
|
|||||||
golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||||
golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
|
golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
|
||||||
golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
|
golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
|
||||||
|
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||||
|
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||||
|
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||||
modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ=
|
modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ=
|
||||||
modernc.org/cc/v4 v4.21.4/go.mod h1:HM7VJTZbUCR3rV8EYBi9wxnJ0ZBRiGE5OeGXNA0IsLQ=
|
modernc.org/cc/v4 v4.21.4/go.mod h1:HM7VJTZbUCR3rV8EYBi9wxnJ0ZBRiGE5OeGXNA0IsLQ=
|
||||||
modernc.org/ccgo/v4 v4.19.2 h1:lwQZgvboKD0jBwdaeVCTouxhxAyN6iawF3STraAal8Y=
|
modernc.org/ccgo/v4 v4.19.2 h1:lwQZgvboKD0jBwdaeVCTouxhxAyN6iawF3STraAal8Y=
|
||||||
|
|||||||
@@ -0,0 +1,167 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/executus/persona"
|
||||||
|
)
|
||||||
|
|
||||||
|
// personaStore is the SQLite-backed persona.Storage. It stores each Agent as a
|
||||||
|
// JSON blob in `data` with a handful of extracted, indexed columns for the
|
||||||
|
// query methods — so the FULL struct round-trips (no domain↔GORM↔DB field-loss
|
||||||
|
// footgun) while owner/name/webhook/schedule lookups stay indexable.
|
||||||
|
type personaStore struct{ db *sql.DB }
|
||||||
|
|
||||||
|
// Personas returns a durable persona.Storage backed by this database.
|
||||||
|
func (d *DB) Personas() persona.Storage { return &personaStore{db: d.sql} }
|
||||||
|
|
||||||
|
var _ persona.Storage = (*personaStore)(nil)
|
||||||
|
|
||||||
|
func (s *personaStore) InitializeAgentStorage(ctx context.Context) error {
|
||||||
|
_, err := s.db.ExecContext(ctx, `
|
||||||
|
CREATE TABLE IF NOT EXISTS agents (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
owner_id TEXT NOT NULL DEFAULT '',
|
||||||
|
name TEXT NOT NULL DEFAULT '',
|
||||||
|
webhook_secret TEXT NOT NULL DEFAULT '',
|
||||||
|
chatbot_channel_filter TEXT NOT NULL DEFAULT '',
|
||||||
|
schedule TEXT NOT NULL DEFAULT '',
|
||||||
|
next_run_at INTEGER NOT NULL DEFAULT 0, -- unix seconds; 0 = unset
|
||||||
|
data TEXT NOT NULL -- full Agent as JSON
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_agents_owner ON agents(owner_id);
|
||||||
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_agents_owner_name ON agents(owner_id, name);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_agents_sched ON agents(schedule, next_run_at);`)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("personaStore.Initialize: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) SaveAgent(ctx context.Context, a *persona.Agent) error {
|
||||||
|
blob, err := json.Marshal(a)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("personaStore.SaveAgent: marshal: %w", err)
|
||||||
|
}
|
||||||
|
var next int64
|
||||||
|
if a.NextRunAt != nil && !a.NextRunAt.IsZero() {
|
||||||
|
next = a.NextRunAt.Unix()
|
||||||
|
}
|
||||||
|
_, err = s.db.ExecContext(ctx, `
|
||||||
|
INSERT INTO agents (id, owner_id, name, webhook_secret, chatbot_channel_filter, schedule, next_run_at, data)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
ON CONFLICT(id) DO UPDATE SET
|
||||||
|
owner_id=excluded.owner_id, name=excluded.name, webhook_secret=excluded.webhook_secret,
|
||||||
|
chatbot_channel_filter=excluded.chatbot_channel_filter, schedule=excluded.schedule,
|
||||||
|
next_run_at=excluded.next_run_at, data=excluded.data`,
|
||||||
|
a.ID, a.OwnerID, a.Name, a.WebhookSecret, a.ChatbotChannelFilter, a.Schedule, next, string(blob))
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("personaStore.SaveAgent: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// scanAgents unmarshals the `data` column of every row in rows.
|
||||||
|
func scanAgents(rows *sql.Rows) ([]*persona.Agent, error) {
|
||||||
|
defer rows.Close()
|
||||||
|
var out []*persona.Agent
|
||||||
|
for rows.Next() {
|
||||||
|
var blob string
|
||||||
|
if err := rows.Scan(&blob); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var a persona.Agent
|
||||||
|
if err := json.Unmarshal([]byte(blob), &a); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, &a)
|
||||||
|
}
|
||||||
|
return out, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) getOne(ctx context.Context, where string, arg ...any) (*persona.Agent, error) {
|
||||||
|
var blob string
|
||||||
|
err := s.db.QueryRowContext(ctx, `SELECT data FROM agents WHERE `+where, arg...).Scan(&blob)
|
||||||
|
switch {
|
||||||
|
case errors.Is(err, sql.ErrNoRows):
|
||||||
|
return nil, persona.ErrNotFound
|
||||||
|
case err != nil:
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
var a persona.Agent
|
||||||
|
if err := json.Unmarshal([]byte(blob), &a); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return &a, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) GetAgent(ctx context.Context, id string) (*persona.Agent, error) {
|
||||||
|
return s.getOne(ctx, "id = ?", id)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) GetAgentByName(ctx context.Context, ownerID, name string) (*persona.Agent, error) {
|
||||||
|
return s.getOne(ctx, "owner_id = ? AND name = ?", ownerID, name)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) GetAgentByWebhookSecret(ctx context.Context, secret string) (*persona.Agent, error) {
|
||||||
|
if secret == "" {
|
||||||
|
return nil, persona.ErrNotFound
|
||||||
|
}
|
||||||
|
return s.getOne(ctx, "webhook_secret = ?", secret)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) ListAgents(ctx context.Context, ownerID string) ([]*persona.Agent, error) {
|
||||||
|
rows, err := s.db.QueryContext(ctx, `SELECT data FROM agents WHERE owner_id = ? ORDER BY name`, ownerID)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("personaStore.ListAgents: %w", err)
|
||||||
|
}
|
||||||
|
return scanAgents(rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) ListAllAgents(ctx context.Context) ([]*persona.Agent, error) {
|
||||||
|
rows, err := s.db.QueryContext(ctx, `SELECT data FROM agents ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("personaStore.ListAllAgents: %w", err)
|
||||||
|
}
|
||||||
|
return scanAgents(rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) DeleteAgent(ctx context.Context, id string) error {
|
||||||
|
if _, err := s.db.ExecContext(ctx, `DELETE FROM agents WHERE id = ?`, id); err != nil {
|
||||||
|
return fmt.Errorf("personaStore.DeleteAgent: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) ListAgentsByChatbotChannelFilter(ctx context.Context) ([]*persona.Agent, error) {
|
||||||
|
rows, err := s.db.QueryContext(ctx, `SELECT data FROM agents WHERE chatbot_channel_filter != '' ORDER BY name`)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("personaStore.ListAgentsByChatbotChannelFilter: %w", err)
|
||||||
|
}
|
||||||
|
return scanAgents(rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) ListScheduledAgents(ctx context.Context, dueBefore time.Time) ([]*persona.Agent, error) {
|
||||||
|
rows, err := s.db.QueryContext(ctx,
|
||||||
|
`SELECT data FROM agents WHERE schedule != '' AND next_run_at > 0 AND next_run_at <= ? ORDER BY next_run_at`,
|
||||||
|
dueBefore.Unix())
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("personaStore.ListScheduledAgents: %w", err)
|
||||||
|
}
|
||||||
|
return scanAgents(rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *personaStore) MarkAgentScheduledRun(ctx context.Context, agentID string, ranAt, nextAt time.Time) error {
|
||||||
|
a, err := s.GetAgent(ctx, agentID)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
a.LastScheduledRunAt = &ranAt
|
||||||
|
a.NextRunAt = &nextAt
|
||||||
|
return s.SaveAgent(ctx, a)
|
||||||
|
}
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.stevedudenhoeffer.com/steve/executus/persona"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestSQLitePersonaStore(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
db, err := Open(":memory:")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
st := db.Personas()
|
||||||
|
if err := st.InitializeAgentStorage(ctx); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Full struct round-trips through the JSON blob (incl. nested + map fields).
|
||||||
|
a := &persona.Agent{
|
||||||
|
ID: "a1", Name: "helper", OwnerID: "o1", SystemPrompt: "be nice",
|
||||||
|
ModelTier: "fast", SkillPalette: []string{"animate"},
|
||||||
|
StateReactEmoji: map[string]string{"running": "⏳"},
|
||||||
|
ChatbotChannelFilter: "general",
|
||||||
|
}
|
||||||
|
if err := st.SaveAgent(ctx, a); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
got, err := st.GetAgent(ctx, "a1")
|
||||||
|
if err != nil || got.SystemPrompt != "be nice" || len(got.SkillPalette) != 1 ||
|
||||||
|
got.StateReactEmoji["running"] != "⏳" {
|
||||||
|
t.Fatalf("round-trip lost fields: %+v (err %v)", got, err)
|
||||||
|
}
|
||||||
|
if byName, err := st.GetAgentByName(ctx, "o1", "helper"); err != nil || byName.ID != "a1" {
|
||||||
|
t.Fatalf("GetAgentByName: %v %+v", err, byName)
|
||||||
|
}
|
||||||
|
if cf, _ := st.ListAgentsByChatbotChannelFilter(ctx); len(cf) != 1 {
|
||||||
|
t.Errorf("ListAgentsByChatbotChannelFilter = %d, want 1", len(cf))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Scheduling: due query + MarkAgentScheduledRun round-trip.
|
||||||
|
now := time.Now().UTC()
|
||||||
|
sched := &persona.Agent{ID: "s1", Name: "cron", OwnerID: "o1", Schedule: "0 * * * *"}
|
||||||
|
due := now.Add(-time.Minute)
|
||||||
|
sched.NextRunAt = &due
|
||||||
|
if err := st.SaveAgent(ctx, sched); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
dueList, _ := st.ListScheduledAgents(ctx, now)
|
||||||
|
if len(dueList) != 1 || dueList[0].ID != "s1" {
|
||||||
|
t.Fatalf("ListScheduledAgents = %+v", dueList)
|
||||||
|
}
|
||||||
|
next := now.Add(time.Hour)
|
||||||
|
if err := st.MarkAgentScheduledRun(ctx, "s1", now, next); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if again, _ := st.ListScheduledAgents(ctx, now); len(again) != 0 {
|
||||||
|
t.Errorf("after MarkAgentScheduledRun, nothing should be due before now: %+v", again)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := st.DeleteAgent(ctx, "a1"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := st.GetAgent(ctx, "a1"); err != persona.ErrNotFound {
|
||||||
|
t.Errorf("GetAgent after delete = %v, want ErrNotFound", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user