summaryrefslogtreecommitdiff
path: root/internal/storage
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
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')
-rw-r--r--internal/storage/db.go219
-rw-r--r--internal/storage/db_test.go38
-rw-r--r--internal/storage/events.go120
-rw-r--r--internal/storage/events_test.go456
4 files changed, 804 insertions, 29 deletions
diff --git a/internal/storage/db.go b/internal/storage/db.go
index 4adc1ba..e03a902 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -7,6 +7,7 @@ import (
"strings"
"time"
+ "github.com/thepeterstone/claudomator/internal/event"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -139,6 +140,21 @@ 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`,
+ `ALTER TABLE executions ADD COLUMN agent TEXT`,
+ `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 {
@@ -175,15 +191,41 @@ func (s *DB) CreateTask(t *task.Task) error {
return fmt.Errorf("marshaling depends_on: %w", err)
}
- _, err = s.db.Exec(`
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback() //nolint:errcheck
+
+ if _, err = tx.Exec(`
INSERT INTO tasks (id, name, description, elaboration_input, project, repository_url, config_json, priority, timeout_ns, retry_json, tags_json, depends_on_json, parent_task_id, state, created_at, updated_at, story_id, acceptance_criteria, checker_for_task_id, checker_report)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
t.ID, t.Name, t.Description, t.ElaborationInput, t.Project, t.RepositoryURL, string(configJSON), string(t.Priority),
t.Timeout.Duration.Nanoseconds(), string(retryJSON), string(tagsJSON), string(depsJSON),
t.ParentTaskID, string(t.State), t.CreatedAt.UTC(), t.UpdatedAt.UTC(), t.StoryID,
t.AcceptanceCriteria, t.CheckerForTaskID, t.CheckerReport,
- )
- return err
+ ); err != nil {
+ return err
+ }
+
+ // A subtask records a subtask_spawned event on its parent's stream so the
+ // parent's history shows the children it dispatched.
+ if t.ParentTaskID != "" {
+ payload, _ := json.Marshal(struct {
+ SubtaskID string `json:"subtask_id"`
+ Name string `json:"name"`
+ }{SubtaskID: t.ID, Name: t.Name})
+ if err = createEventTx(tx, &event.Event{
+ TaskID: t.ParentTaskID,
+ Kind: event.KindSubtaskSpawned,
+ Actor: event.ActorAgent,
+ Payload: payload,
+ }); err != nil {
+ return err
+ }
+ }
+
+ return tx.Commit()
}
// GetTask retrieves a task by ID.
@@ -247,8 +289,17 @@ func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) {
return tasks, rows.Err()
}
-// UpdateTaskState atomically updates a task's state, enforcing valid transitions.
+// UpdateTaskState atomically updates a task's state, enforcing valid
+// transitions. The transition is attributed to the system actor; callers
+// acting on behalf of a user should use UpdateTaskStateBy.
func (s *DB) UpdateTaskState(id string, newState task.State) error {
+ return s.UpdateTaskStateBy(id, newState, event.ActorSystem)
+}
+
+// UpdateTaskStateBy is UpdateTaskState with explicit actor attribution. It
+// writes a state_change event in the same transaction as the row update so the
+// event stream and task state never diverge.
+func (s *DB) UpdateTaskStateBy(id string, newState task.State, actor event.Actor) error {
tx, err := s.db.Begin()
if err != nil {
return err
@@ -271,6 +322,14 @@ func (s *DB) UpdateTaskState(id string, newState task.State) error {
if _, err := tx.Exec(`UPDATE tasks SET state = ?, updated_at = ? WHERE id = ?`, string(newState), now, id); err != nil {
return err
}
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: actor,
+ Payload: stateChangePayload(task.State(currentState), newState),
+ }); err != nil {
+ return err
+ }
return tx.Commit()
}
@@ -305,6 +364,15 @@ func (s *DB) ResetTaskForRetry(id string) (*task.Task, error) {
return nil, err
}
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: event.ActorUser,
+ Payload: stateChangePayload(t.State, task.StateQueued),
+ }); err != nil {
+ return nil, err
+ }
+
if err := tx.Commit(); err != nil {
return nil, err
}
@@ -326,20 +394,35 @@ func (s *DB) UpdateTaskAgent(id string, agent task.AgentConfig) error {
// RejectTask sets a task's state to PENDING and stores the rejection comment.
func (s *DB) RejectTask(id, comment string) error {
- now := time.Now().UTC()
- result, err := s.db.Exec(`UPDATE tasks SET state = ?, rejection_comment = ?, updated_at = ? WHERE id = ?`,
- string(task.StatePending), comment, now, id)
+ tx, err := s.db.Begin()
if err != nil {
return err
}
- n, err := result.RowsAffected()
- if err != nil {
+ defer tx.Rollback() //nolint:errcheck
+
+ var currentState string
+ if err := tx.QueryRow(`SELECT state FROM tasks WHERE id = ?`, id).Scan(&currentState); err != nil {
+ if err == sql.ErrNoRows {
+ return fmt.Errorf("task %q not found", id)
+ }
return err
}
- if n == 0 {
- return fmt.Errorf("task %q not found", id)
+
+ now := time.Now().UTC()
+ if _, err := tx.Exec(`UPDATE tasks SET state = ?, rejection_comment = ?, updated_at = ? WHERE id = ?`,
+ string(task.StatePending), comment, now, id); err != nil {
+ return err
}
- return nil
+
+ if err := createEventTx(tx, &event.Event{
+ TaskID: id,
+ Kind: event.KindStateChange,
+ Actor: event.ActorUser,
+ Payload: stateChangePayload(task.State(currentState), task.StatePending),
+ }); err != nil {
+ return err
+ }
+ return tx.Commit()
}
// TaskUpdate holds the fields that UpdateTask may change.
@@ -454,6 +537,7 @@ type Execution struct {
ErrorMsg string
SessionID string // claude --session-id; persisted for resume
SandboxDir string // preserved sandbox path when task is BLOCKED; resume must run here
+ Agent string // provider that ran this execution (claude/gemini/local); for budget accounting
Changestats *task.Changestats // stored as JSON; nil if not yet recorded
Commits []task.GitCommit // stored as JSON; empty if no commits
@@ -502,10 +586,10 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error {
commitsJSON = string(b)
}
if _, err := tx.Exec(`
- INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)`,
+ INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, agent)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`,
e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status,
- e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON,
+ e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, e.Agent,
); err != nil {
return err
}
@@ -519,6 +603,32 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error {
return tx.Commit()
}
+// SpendByProviderSince returns total cost_usd per provider (executions.agent)
+// for executions started at or after `since`. Executions with no recorded agent
+// are grouped under the empty string and can be ignored by callers.
+func (s *DB) SpendByProviderSince(since time.Time) (map[string]float64, error) {
+ rows, err := s.db.Query(`
+ SELECT COALESCE(agent, ''), COALESCE(SUM(cost_usd), 0)
+ FROM executions
+ WHERE start_time >= ?
+ GROUP BY agent`, since.UTC())
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ out := make(map[string]float64)
+ for rows.Next() {
+ var agent string
+ var cost float64
+ if err := rows.Scan(&agent, &cost); err != nil {
+ return nil, err
+ }
+ out[agent] = cost
+ }
+ return out, rows.Err()
+}
+
// CreateExecution inserts an execution record.
func (s *DB) CreateExecution(e *Execution) error {
var changestatsJSON *string
@@ -539,23 +649,23 @@ func (s *DB) CreateExecution(e *Execution) error {
commitsJSON = string(b)
}
_, err := s.db.Exec(`
- INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+ INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status,
- e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut,
+ e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, e.Agent,
)
return err
}
// GetExecution retrieves an execution by ID.
func (s *DB) GetExecution(id string) (*Execution, error) {
- row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE id = ?`, id)
+ row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE id = ?`, id)
return scanExecution(row)
}
// ListExecutions returns executions for a task.
func (s *DB) ListExecutions(taskID string) ([]*Execution, error) {
- rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID)
+ rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID)
if err != nil {
return nil, err
}
@@ -574,7 +684,7 @@ func (s *DB) ListExecutions(taskID string) ([]*Execution, error) {
// GetLatestExecution returns the most recent execution for a task.
func (s *DB) GetLatestExecution(taskID string) (*Execution, error) {
- row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID)
+ row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID)
return scanExecution(row)
}
@@ -824,17 +934,57 @@ func (s *DB) ListRecentExecutions(since time.Time, limit int, taskID string) ([]
// UpdateTaskQuestion stores the pending question JSON on a task.
// Pass empty string to clear the question after it has been answered.
+// A non-empty question additionally emits a clarification_request event
+// (the agent asking for input); clearing emits nothing.
func (s *DB) UpdateTaskQuestion(taskID, questionJSON string) error {
- _, err := s.db.Exec(`UPDATE tasks SET question_json = ?, updated_at = ? WHERE id = ?`,
- questionJSON, time.Now().UTC(), taskID)
- return err
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback() //nolint:errcheck
+
+ if _, err := tx.Exec(`UPDATE tasks SET question_json = ?, updated_at = ? WHERE id = ?`,
+ questionJSON, time.Now().UTC(), taskID); err != nil {
+ return err
+ }
+ if questionJSON != "" {
+ if err := createEventTx(tx, &event.Event{
+ TaskID: taskID,
+ Kind: event.KindClarificationRequest,
+ Actor: event.ActorAgent,
+ Payload: json.RawMessage(questionJSON),
+ }); err != nil {
+ return err
+ }
+ }
+ return tx.Commit()
}
-// UpdateTaskSummary stores the agent's final summary paragraph on a task.
+// UpdateTaskSummary stores the agent's final summary paragraph on a task and
+// emits a summary event.
func (s *DB) UpdateTaskSummary(taskID, summary string) error {
- _, err := s.db.Exec(`UPDATE tasks SET summary = ?, updated_at = ? WHERE id = ?`,
- summary, time.Now().UTC(), taskID)
- return err
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback() //nolint:errcheck
+
+ if _, err := tx.Exec(`UPDATE tasks SET summary = ?, updated_at = ? WHERE id = ?`,
+ summary, time.Now().UTC(), taskID); err != nil {
+ return err
+ }
+ payload, _ := json.Marshal(struct {
+ Text string `json:"text"`
+ }{Text: summary})
+ if err := createEventTx(tx, &event.Event{
+ TaskID: taskID,
+ Kind: event.KindSummary,
+ Actor: event.ActorAgent,
+ Payload: payload,
+ }); err != nil {
+ return err
+ }
+ return tx.Commit()
}
// UpdateTaskCheckerReport sets the checker_report field on a task.
@@ -887,6 +1037,15 @@ func (s *DB) AppendTaskInteraction(taskID string, interaction task.Interaction)
string(updated), time.Now().UTC(), taskID); err != nil {
return err
}
+ payload, _ := json.Marshal(interaction)
+ if err := createEventTx(tx, &event.Event{
+ TaskID: taskID,
+ Kind: event.KindClarificationAnswer,
+ Actor: event.ActorUser,
+ Payload: payload,
+ }); err != nil {
+ return err
+ }
return tx.Commit()
}
@@ -1006,8 +1165,9 @@ func scanExecution(row scanner) (*Execution, error) {
var commitsJSON sql.NullString
var tokensIn sql.NullInt64
var tokensOut sql.NullInt64
+ var agent sql.NullString
err := row.Scan(&e.ID, &e.TaskID, &e.StartTime, &e.EndTime, &e.ExitCode, &e.Status,
- &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut)
+ &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut, &agent)
if err != nil {
return nil, err
}
@@ -1015,6 +1175,7 @@ func scanExecution(row scanner) (*Execution, error) {
e.SandboxDir = sandboxDir.String
e.TokensIn = tokensIn.Int64
e.TokensOut = tokensOut.Int64
+ e.Agent = agent.String
if changestatsJSON.Valid && changestatsJSON.String != "" {
var cs task.Changestats
if err := json.Unmarshal([]byte(changestatsJSON.String), &cs); err != nil {
diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go
index 0e67e02..4d744a4 100644
--- a/internal/storage/db_test.go
+++ b/internal/storage/db_test.go
@@ -309,6 +309,44 @@ func TestListTasks_WithLimit(t *testing.T) {
}
}
+func TestSpendByProviderSince(t *testing.T) {
+ db := testDB(t)
+ now := time.Now().UTC()
+
+ tk := &task.Task{
+ ID: "spend-task", Name: "S", Agent: task.AgentConfig{Type: "claude", Instructions: "x"},
+ Priority: task.PriorityNormal, Retry: task.RetryConfig{MaxAttempts: 1, Backoff: "linear"},
+ Tags: []string{}, DependsOn: []string{}, State: task.StatePending, CreatedAt: now, UpdatedAt: now,
+ }
+ if err := db.CreateTask(tk); err != nil {
+ t.Fatal(err)
+ }
+
+ // Two recent claude execs, one recent gemini, one old claude (outside window).
+ execs := []*Execution{
+ {ID: "s1", TaskID: "spend-task", StartTime: now.Add(-1 * time.Hour), Status: "COMPLETED", CostUSD: 1.5, Agent: "claude"},
+ {ID: "s2", TaskID: "spend-task", StartTime: now.Add(-2 * time.Hour), Status: "COMPLETED", CostUSD: 2.0, Agent: "claude"},
+ {ID: "s3", TaskID: "spend-task", StartTime: now.Add(-30 * time.Minute), Status: "COMPLETED", CostUSD: 0.75, Agent: "gemini"},
+ {ID: "s4", TaskID: "spend-task", StartTime: now.Add(-10 * time.Hour), Status: "COMPLETED", CostUSD: 99, Agent: "claude"},
+ }
+ for _, e := range execs {
+ if err := db.CreateExecution(e); err != nil {
+ t.Fatalf("create exec %s: %v", e.ID, err)
+ }
+ }
+
+ spends, err := db.SpendByProviderSince(now.Add(-5 * time.Hour))
+ if err != nil {
+ t.Fatalf("SpendByProviderSince: %v", err)
+ }
+ if got := spends["claude"]; got != 3.5 {
+ t.Errorf("claude spend: want 3.5 (old $99 excluded), got %v", got)
+ }
+ if got := spends["gemini"]; got != 0.75 {
+ t.Errorf("gemini spend: want 0.75, got %v", got)
+ }
+}
+
func TestCreateExecution_AndGet(t *testing.T) {
db := testDB(t)
now := time.Now().UTC().Truncate(time.Second)
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
+}
diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go
new file mode 100644
index 0000000..c5d3c37
--- /dev/null
+++ b/internal/storage/events_test.go
@@ -0,0 +1,456 @@
+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 TestUpdateTaskState_WritesStateChangeEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1") // starts PENDING
+
+ if err := db.UpdateTaskState("task-1", task.StateQueued); err != nil {
+ t.Fatalf("UpdateTaskState: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ e := events[0]
+ if e.Kind != event.KindStateChange {
+ t.Errorf("kind: got %v want state_change", e.Kind)
+ }
+ if e.Actor != event.ActorSystem {
+ t.Errorf("actor: got %v want system", e.Actor)
+ }
+ var payload struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }
+ if err := json.Unmarshal(e.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal payload: %v", err)
+ }
+ if payload.From != task.StatePending || payload.To != task.StateQueued {
+ t.Errorf("payload: got %s→%s want PENDING→QUEUED", payload.From, payload.To)
+ }
+}
+
+func TestUpdateTaskStateBy_AttributesActor(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ if err := db.UpdateTaskStateBy("task-1", task.StateCancelled, event.ActorUser); err != nil {
+ t.Fatalf("UpdateTaskStateBy: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ if events[0].Actor != event.ActorUser {
+ t.Errorf("actor: got %v want user", events[0].Actor)
+ }
+}
+
+func TestUpdateTaskState_InvalidTransition_WritesNoEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1") // PENDING
+
+ // PENDING → COMPLETED is not a valid transition.
+ if err := db.UpdateTaskState("task-1", task.StateCompleted); err == nil {
+ t.Fatal("expected invalid transition error, got nil")
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 0 {
+ t.Fatalf("expected no events after rejected transition, got %d", len(events))
+ }
+}
+
+func TestRejectTask_WritesUserStateChangeEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+ // Drive PENDING → QUEUED → RUNNING → READY so reject is valid.
+ for _, st := range []task.State{task.StateQueued, task.StateRunning, task.StateReady} {
+ if err := db.UpdateTaskState("task-1", st); err != nil {
+ t.Fatalf("transition to %s: %v", st, err)
+ }
+ }
+
+ if err := db.RejectTask("task-1", "needs work"); err != nil {
+ t.Fatalf("RejectTask: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ last := events[len(events)-1]
+ if last.Kind != event.KindStateChange || last.Actor != event.ActorUser {
+ t.Errorf("reject event: got kind=%v actor=%v want state_change/user", last.Kind, last.Actor)
+ }
+ var payload struct {
+ From task.State `json:"from"`
+ To task.State `json:"to"`
+ }
+ if err := json.Unmarshal(last.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if payload.From != task.StateReady || payload.To != task.StatePending {
+ t.Errorf("payload: got %s→%s want READY→PENDING", payload.From, payload.To)
+ }
+}
+
+func TestUpdateTaskQuestion_EmitsClarificationRequest(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ q := `{"text":"which env?","options":["dev","prod"]}`
+ if err := db.UpdateTaskQuestion("task-1", q); err != nil {
+ t.Fatalf("UpdateTaskQuestion: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ e := events[0]
+ if e.Kind != event.KindClarificationRequest || e.Actor != event.ActorAgent {
+ t.Errorf("got kind=%v actor=%v want clarification_request/agent", e.Kind, e.Actor)
+ }
+ if string(e.Payload) != q {
+ t.Errorf("payload: got %s want %s", e.Payload, q)
+ }
+}
+
+func TestUpdateTaskQuestion_ClearEmitsNothing(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ if err := db.UpdateTaskQuestion("task-1", ""); err != nil {
+ t.Fatalf("UpdateTaskQuestion clear: %v", err)
+ }
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 0 {
+ t.Fatalf("clearing question should emit no event, got %d", len(events))
+ }
+}
+
+func TestUpdateTaskSummary_EmitsSummaryEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ if err := db.UpdateTaskSummary("task-1", "Did the thing."); err != nil {
+ t.Fatalf("UpdateTaskSummary: %v", err)
+ }
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ e := events[0]
+ if e.Kind != event.KindSummary || e.Actor != event.ActorAgent {
+ t.Errorf("got kind=%v actor=%v want summary/agent", e.Kind, e.Actor)
+ }
+ var payload struct {
+ Text string `json:"text"`
+ }
+ if err := json.Unmarshal(e.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if payload.Text != "Did the thing." {
+ t.Errorf("payload text: got %q want %q", payload.Text, "Did the thing.")
+ }
+}
+
+func TestAppendTaskInteraction_EmitsClarificationAnswer(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "task-1")
+
+ interaction := task.Interaction{
+ QuestionText: "which env?",
+ Options: []string{"dev", "prod"},
+ Answer: "prod",
+ }
+ if err := db.AppendTaskInteraction("task-1", interaction); err != nil {
+ t.Fatalf("AppendTaskInteraction: %v", err)
+ }
+
+ events, err := db.ListEvents("task-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 1 {
+ t.Fatalf("expected 1 event, got %d", len(events))
+ }
+ e := events[0]
+ if e.Kind != event.KindClarificationAnswer || e.Actor != event.ActorUser {
+ t.Errorf("got kind=%v actor=%v want clarification_answer/user", e.Kind, e.Actor)
+ }
+ var payload task.Interaction
+ if err := json.Unmarshal(e.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if payload.Answer != "prod" || payload.QuestionText != "which env?" {
+ t.Errorf("payload: got %+v", payload)
+ }
+}
+
+func TestCreateTask_Subtask_EmitsSubtaskSpawnedOnParent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "parent-1")
+
+ now := time.Now().UTC().Truncate(time.Second)
+ child := &task.Task{
+ ID: "child-1",
+ Name: "Do the subthing",
+ ParentTaskID: "parent-1",
+ Agent: task.AgentConfig{Type: "claude", Instructions: "noop"},
+ Priority: task.PriorityNormal,
+ State: task.StatePending,
+ CreatedAt: now,
+ UpdatedAt: now,
+ }
+ if err := db.CreateTask(child); err != nil {
+ t.Fatalf("CreateTask child: %v", err)
+ }
+
+ // Parent should have the subtask_spawned event.
+ parentEvents, err := db.ListEvents("parent-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents parent: %v", err)
+ }
+ if len(parentEvents) != 1 {
+ t.Fatalf("expected 1 event on parent, got %d", len(parentEvents))
+ }
+ e := parentEvents[0]
+ if e.Kind != event.KindSubtaskSpawned || e.Actor != event.ActorAgent {
+ t.Errorf("got kind=%v actor=%v want subtask_spawned/agent", e.Kind, e.Actor)
+ }
+ var payload struct {
+ SubtaskID string `json:"subtask_id"`
+ Name string `json:"name"`
+ }
+ if err := json.Unmarshal(e.Payload, &payload); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if payload.SubtaskID != "child-1" || payload.Name != "Do the subthing" {
+ t.Errorf("payload: got %+v", payload)
+ }
+
+ // Child should have no events of its own from creation.
+ childEvents, err := db.ListEvents("child-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents child: %v", err)
+ }
+ if len(childEvents) != 0 {
+ t.Errorf("expected no events on child, got %d", len(childEvents))
+ }
+}
+
+func TestCreateTask_TopLevel_EmitsNoEvent(t *testing.T) {
+ db := testDB(t)
+ makeTaskForEvent(t, db, "top-1") // no parent
+
+ events, err := db.ListEvents("top-1", 0)
+ if err != nil {
+ t.Fatalf("ListEvents: %v", err)
+ }
+ if len(events) != 0 {
+ t.Errorf("top-level task creation should emit no event, got %d", len(events))
+ }
+}
+
+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")
+ }
+ })
+ }
+}