From 39cc108b5f0bf3d0238a1707d657a77b7a3df184 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 09:08:52 +0000 Subject: feat(event,storage): add events table substrate for OSS redesign New internal/event package defines the Event type with canonical Kind and Actor constants. New events table is appended via additive migration (id, task_id, seq, ts, kind, actor, payload_json) with seq as per-task monotonic, assigned by the storage layer inside a transaction. CreateEvent + ListEvents support the future migration of question_json, interactions_json, summary, and elaboration_input off the tasks table into a unified, replayable event stream. No existing code paths are modified yet; this is pure substrate for Phase 1. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/storage/db.go | 14 ++++ internal/storage/events.go | 99 ++++++++++++++++++++++ internal/storage/events_test.go | 182 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 295 insertions(+) create mode 100644 internal/storage/events.go create mode 100644 internal/storage/events_test.go (limited to 'internal/storage') diff --git a/internal/storage/db.go b/internal/storage/db.go index 4adc1ba..3a4f059 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -139,6 +139,20 @@ func (s *DB) migrate() error { `ALTER TABLE tasks ADD COLUMN checker_report TEXT NOT NULL DEFAULT ''`, `ALTER TABLE executions ADD COLUMN tokens_in INTEGER`, `ALTER TABLE executions ADD COLUMN tokens_out INTEGER`, + `CREATE TABLE IF NOT EXISTS events ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + seq INTEGER NOT NULL, + ts DATETIME NOT NULL, + kind TEXT NOT NULL, + actor TEXT NOT NULL, + payload_json TEXT NOT NULL DEFAULT '{}', + UNIQUE(task_id, seq), + FOREIGN KEY (task_id) REFERENCES tasks(id) + )`, + `CREATE INDEX IF NOT EXISTS idx_events_task_id_seq ON events(task_id, seq)`, + `CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts)`, + `CREATE INDEX IF NOT EXISTS idx_events_kind ON events(kind)`, } for _, m := range migrations { if _, err := s.db.Exec(m); err != nil { diff --git a/internal/storage/events.go b/internal/storage/events.go new file mode 100644 index 0000000..82016ab --- /dev/null +++ b/internal/storage/events.go @@ -0,0 +1,99 @@ +package storage + +import ( + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/thepeterstone/claudomator/internal/event" +) + +// CreateEvent appends an event to a task's stream. Seq is assigned by the +// database as MAX(seq)+1 within a transaction. On return, e.ID, e.Seq, and +// e.Timestamp are populated (timestamps default to time.Now().UTC()). +func (s *DB) CreateEvent(e *event.Event) error { + if e == nil { + return errors.New("event is nil") + } + if e.TaskID == "" { + return errors.New("event.TaskID required") + } + if e.Kind == "" { + return errors.New("event.Kind required") + } + if e.Actor == "" { + return errors.New("event.Actor required") + } + if len(e.Payload) == 0 { + e.Payload = json.RawMessage("{}") + } + if e.ID == "" { + e.ID = uuid.NewString() + } + if e.Timestamp.IsZero() { + e.Timestamp = time.Now().UTC() + } else { + e.Timestamp = e.Timestamp.UTC() + } + + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + var nextSeq int64 + if err := tx.QueryRow(`SELECT COALESCE(MAX(seq), 0) + 1 FROM events WHERE task_id = ?`, e.TaskID).Scan(&nextSeq); err != nil { + return fmt.Errorf("computing next seq: %w", err) + } + e.Seq = nextSeq + + if _, err := tx.Exec( + `INSERT INTO events (id, task_id, seq, ts, kind, actor, payload_json) VALUES (?, ?, ?, ?, ?, ?, ?)`, + e.ID, e.TaskID, e.Seq, e.Timestamp, string(e.Kind), string(e.Actor), string(e.Payload), + ); err != nil { + return fmt.Errorf("inserting event: %w", err) + } + return tx.Commit() +} + +// ListEvents returns events for a task in seq order. Pass sinceSeq=0 for the +// full history, or a higher value to fetch only events strictly after that seq +// (for incremental polling / SSE catch-up). +func (s *DB) ListEvents(taskID string, sinceSeq int64) ([]*event.Event, error) { + rows, err := s.db.Query( + `SELECT id, task_id, seq, ts, kind, actor, payload_json FROM events WHERE task_id = ? AND seq > ? ORDER BY seq ASC`, + taskID, sinceSeq, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var out []*event.Event + for rows.Next() { + e, err := scanEvent(rows) + if err != nil { + return nil, err + } + out = append(out, e) + } + return out, rows.Err() +} + +func scanEvent(row scanner) (*event.Event, error) { + var ( + e event.Event + kind string + actor string + payload string + ) + if err := row.Scan(&e.ID, &e.TaskID, &e.Seq, &e.Timestamp, &kind, &actor, &payload); err != nil { + return nil, err + } + e.Kind = event.Kind(kind) + e.Actor = event.Actor(actor) + e.Payload = json.RawMessage(payload) + return &e, nil +} diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go new file mode 100644 index 0000000..3b342c7 --- /dev/null +++ b/internal/storage/events_test.go @@ -0,0 +1,182 @@ +package storage + +import ( + "encoding/json" + "testing" + "time" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/task" +) + +// makeTaskForEvent creates a minimal task row so events can satisfy the FK. +func makeTaskForEvent(t *testing.T, db *DB, id string) { + t.Helper() + now := time.Now().UTC().Truncate(time.Second) + tk := &task.Task{ + ID: id, + Name: id, + Agent: task.AgentConfig{Type: "claude", Instructions: "noop"}, + Priority: task.PriorityNormal, + State: task.StatePending, + CreatedAt: now, + UpdatedAt: now, + } + if err := db.CreateTask(tk); err != nil { + t.Fatalf("creating task: %v", err) + } +} + +func TestCreateEvent_AssignsSeqAndDefaults(t *testing.T) { + db := testDB(t) + makeTaskForEvent(t, db, "task-1") + + e := &event.Event{ + TaskID: "task-1", + Kind: event.KindStateChange, + Actor: event.ActorSystem, + } + if err := db.CreateEvent(e); err != nil { + t.Fatalf("CreateEvent: %v", err) + } + if e.ID == "" { + t.Error("expected ID populated") + } + if e.Seq != 1 { + t.Errorf("expected seq=1, got %d", e.Seq) + } + if e.Timestamp.IsZero() { + t.Error("expected Timestamp populated") + } + if string(e.Payload) != "{}" { + t.Errorf("expected default payload {}, got %s", e.Payload) + } +} + +func TestCreateEvent_SeqIsMonotonicPerTask(t *testing.T) { + db := testDB(t) + makeTaskForEvent(t, db, "task-A") + makeTaskForEvent(t, db, "task-B") + + // 3 events on task-A, 2 on task-B, interleaved. + for _, taskID := range []string{"task-A", "task-A", "task-B", "task-A", "task-B"} { + e := &event.Event{ + TaskID: taskID, + Kind: event.KindAgentMessage, + Actor: event.ActorAgent, + } + if err := db.CreateEvent(e); err != nil { + t.Fatalf("CreateEvent: %v", err) + } + } + + a, err := db.ListEvents("task-A", 0) + if err != nil { + t.Fatalf("ListEvents A: %v", err) + } + if len(a) != 3 { + t.Fatalf("expected 3 events on task-A, got %d", len(a)) + } + for i, e := range a { + want := int64(i + 1) + if e.Seq != want { + t.Errorf("task-A event %d: seq=%d want %d", i, e.Seq, want) + } + } + + b, err := db.ListEvents("task-B", 0) + if err != nil { + t.Fatalf("ListEvents B: %v", err) + } + if len(b) != 2 { + t.Fatalf("expected 2 events on task-B, got %d", len(b)) + } + for i, e := range b { + want := int64(i + 1) + if e.Seq != want { + t.Errorf("task-B event %d: seq=%d want %d", i, e.Seq, want) + } + } +} + +func TestListEvents_SinceSeq(t *testing.T) { + db := testDB(t) + makeTaskForEvent(t, db, "task-1") + + for i := 0; i < 5; i++ { + e := &event.Event{ + TaskID: "task-1", + Kind: event.KindAgentMessage, + Actor: event.ActorAgent, + } + if err := db.CreateEvent(e); err != nil { + t.Fatalf("CreateEvent: %v", err) + } + } + + got, err := db.ListEvents("task-1", 3) + if err != nil { + t.Fatalf("ListEvents: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 events with seq>3, got %d", len(got)) + } + if got[0].Seq != 4 || got[1].Seq != 5 { + t.Errorf("expected seqs 4,5; got %d,%d", got[0].Seq, got[1].Seq) + } +} + +func TestListEvents_PreservesPayloadAndKind(t *testing.T) { + db := testDB(t) + makeTaskForEvent(t, db, "task-1") + + payload := json.RawMessage(`{"question":"which env?","options":["dev","prod"]}`) + written := &event.Event{ + TaskID: "task-1", + Kind: event.KindClarificationRequest, + Actor: event.ActorAgent, + Payload: payload, + } + if err := db.CreateEvent(written); err != nil { + t.Fatalf("CreateEvent: %v", err) + } + + got, err := db.ListEvents("task-1", 0) + if err != nil { + t.Fatalf("ListEvents: %v", err) + } + if len(got) != 1 { + t.Fatalf("expected 1 event, got %d", len(got)) + } + e := got[0] + if e.Kind != event.KindClarificationRequest { + t.Errorf("kind: got %v want %v", e.Kind, event.KindClarificationRequest) + } + if e.Actor != event.ActorAgent { + t.Errorf("actor: got %v want %v", e.Actor, event.ActorAgent) + } + if string(e.Payload) != string(payload) { + t.Errorf("payload: got %s want %s", e.Payload, payload) + } +} + +func TestCreateEvent_RequiresFields(t *testing.T) { + db := testDB(t) + + cases := []struct { + name string + e *event.Event + }{ + {"nil", nil}, + {"empty taskID", &event.Event{Kind: event.KindAgentMessage, Actor: event.ActorAgent}}, + {"empty kind", &event.Event{TaskID: "t", Actor: event.ActorAgent}}, + {"empty actor", &event.Event{TaskID: "t", Kind: event.KindAgentMessage}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if err := db.CreateEvent(tc.e); err == nil { + t.Errorf("expected error, got nil") + } + }) + } +} -- cgit v1.2.3