summaryrefslogtreecommitdiff
path: root/internal/storage/events.go
diff options
context:
space:
mode:
authorPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
committerPeter Stone <thepeterstone@gmail.com>2026-06-03 23:01:39 +0000
commit68cf1e24dd2ca2612b72babc4b35280cd3512f92 (patch)
tree14b73f21570a2f42ae729ca7e9676de6628d6819 /internal/storage/events.go
parentb28cfc6ff288d083f6c8e9c055b69bfcadbceccc (diff)
parent7388337d3be5cc7f65ef547e30b2e39884dd165b (diff)
merge: integrate oss branch — budget gating, MCP back-channel, events, loopback-only bind
Key features from OSS branch: - Budget gating: rolling per-provider spend caps with BUDGET_EXCEEDED task state - Agent MCP back-channel: runners mint tokens; /mcp endpoint resolves them - Chatbot MCP server at /chatbot/mcp (requires api_token in config) - Event timeline: unified event stream replacing ad-hoc question/summary flows - Loopback-only default bind (127.0.0.1:8484); external_bind_allowed=true to expose - repository_url required on task creation (enforces traceability) - Auto-checker: spawns verification task after each execution Conflict resolutions: - serve.go: keep cfg.Runners.XEnabled() conditional registration from main + agentRegistry from oss - config.go: keep RunnersConfig from main + BudgetConfig/ExternalBindAllowed from oss - server.go: handleStaticFiles (base-path rewrite) kept; deduplicated duplicate from both sides - executor/claude.go: accepted oss deletion (ClaudeRunner replaced by ContainerRunner) - container_test.go: added noopChannel + AgentChannel parameter to Run call Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Diffstat (limited to 'internal/storage/events.go')
-rw-r--r--internal/storage/events.go120
1 files changed, 120 insertions, 0 deletions
diff --git a/internal/storage/events.go b/internal/storage/events.go
new file mode 100644
index 0000000..2051127
--- /dev/null
+++ b/internal/storage/events.go
@@ -0,0 +1,120 @@
+package storage
+
+import (
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/task"
+)
+
+// 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 {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if err := createEventTx(tx, e); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// createEventTx validates, defaults, assigns a per-task monotonic seq, and
+// inserts an event using the supplied transaction. The caller commits. This
+// lets state-changing methods write their state_change event atomically with
+// the row update.
+func createEventTx(tx *sql.Tx, 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()
+ }
+
+ 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 nil
+}
+
+// stateChangePayload builds the JSON payload for a state_change event.
+func stateChangePayload(from, to task.State) json.RawMessage {
+ b, _ := json.Marshal(struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }{From: from, To: to})
+ return b
+}
+
+// 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
+}