summaryrefslogtreecommitdiff
path: root/internal
diff options
context:
space:
mode:
Diffstat (limited to 'internal')
-rw-r--r--internal/event/event.go50
-rw-r--r--internal/event/event_test.go42
-rw-r--r--internal/storage/db.go14
-rw-r--r--internal/storage/events.go99
-rw-r--r--internal/storage/events_test.go182
5 files changed, 387 insertions, 0 deletions
diff --git a/internal/event/event.go b/internal/event/event.go
new file mode 100644
index 0000000..2573fde
--- /dev/null
+++ b/internal/event/event.go
@@ -0,0 +1,50 @@
+// Package event defines the Event type and canonical kinds/actors for the
+// task observability stream. Events are immutable, per-task, monotonically
+// sequenced records persisted by internal/storage.
+package event
+
+import (
+ "encoding/json"
+ "time"
+)
+
+// Kind enumerates the canonical event types.
+type Kind string
+
+const (
+ KindStateChange Kind = "state_change"
+ KindAgentMessage Kind = "agent_message"
+ KindClarificationRequest Kind = "clarification_request"
+ KindClarificationAnswer Kind = "clarification_answer"
+ KindHumanMessage Kind = "human_message"
+ KindSummary Kind = "summary"
+ KindCommitMade Kind = "commit_made"
+ KindSubtaskSpawned Kind = "subtask_spawned"
+ KindCostReport Kind = "cost_report"
+ KindExecutionStarted Kind = "execution_started"
+ KindExecutionEnded Kind = "execution_ended"
+)
+
+// Actor identifies the originator of an event.
+type Actor string
+
+const (
+ ActorAgent Actor = "agent"
+ ActorUser Actor = "user"
+ ActorChatbot Actor = "chatbot"
+ ActorSystem Actor = "system"
+ ActorWebhook Actor = "webhook"
+)
+
+// Event is a single timestamped record on a task's audit/observability stream.
+// Once written, events are append-only — the seq column is per-task monotonic
+// and assigned by the storage layer on insert.
+type Event struct {
+ ID string `json:"id"`
+ TaskID string `json:"task_id"`
+ Seq int64 `json:"seq"`
+ Timestamp time.Time `json:"timestamp"`
+ Kind Kind `json:"kind"`
+ Actor Actor `json:"actor"`
+ Payload json.RawMessage `json:"payload"`
+}
diff --git a/internal/event/event_test.go b/internal/event/event_test.go
new file mode 100644
index 0000000..a2cd143
--- /dev/null
+++ b/internal/event/event_test.go
@@ -0,0 +1,42 @@
+package event
+
+import (
+ "encoding/json"
+ "testing"
+ "time"
+)
+
+func TestEvent_JSONRoundTrip(t *testing.T) {
+ original := Event{
+ ID: "evt-1",
+ TaskID: "task-1",
+ Seq: 42,
+ Timestamp: time.Date(2026, 5, 22, 12, 0, 0, 0, time.UTC),
+ Kind: KindClarificationRequest,
+ Actor: ActorAgent,
+ Payload: json.RawMessage(`{"question":"which env?"}`),
+ }
+
+ data, err := json.Marshal(original)
+ if err != nil {
+ t.Fatalf("marshal: %v", err)
+ }
+
+ var got Event
+ if err := json.Unmarshal(data, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+
+ if got.ID != original.ID || got.TaskID != original.TaskID || got.Seq != original.Seq {
+ t.Errorf("id/task/seq mismatch: got %+v want %+v", got, original)
+ }
+ if !got.Timestamp.Equal(original.Timestamp) {
+ t.Errorf("timestamp mismatch: got %v want %v", got.Timestamp, original.Timestamp)
+ }
+ if got.Kind != original.Kind || got.Actor != original.Actor {
+ t.Errorf("kind/actor mismatch: got %v/%v want %v/%v", got.Kind, got.Actor, original.Kind, original.Actor)
+ }
+ if string(got.Payload) != string(original.Payload) {
+ t.Errorf("payload mismatch: got %s want %s", got.Payload, original.Payload)
+ }
+}
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")
+ }
+ })
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 10:33:51 +0000