From dc97d7ae00bd25668ba495fba249ceade0c6b100 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 08:07:50 +0000 Subject: feat(storage): emit state_change events on task transitions UpdateTaskState now writes a state_change event in the same transaction as the row update, so the event stream and task state never diverge. Adds UpdateTaskStateBy(id, newState, actor) for explicit actor attribution; UpdateTaskState delegates with ActorSystem (correct for the executor, which drives the state machine). RejectTask and ResetTaskForRetry also emit state_change events attributed to the user. API accept and cancel handlers now attribute their transitions to the user via UpdateTaskStateBy. The executor's Store interface is unchanged, so no test doubles needed updating. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/api/server.go | 5 +- internal/storage/db.go | 60 +++++++++++++++++++---- internal/storage/events.go | 35 ++++++++++--- internal/storage/events_test.go | 106 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 188 insertions(+), 18 deletions(-) (limited to 'internal') diff --git a/internal/api/server.go b/internal/api/server.go index 28cfe4a..ff3a111 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -11,6 +11,7 @@ import ( "time" "github.com/thepeterstone/claudomator/internal/config" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/executor" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/notify" @@ -285,7 +286,7 @@ func (s *Server) handleCancelTask(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusConflict, map[string]string{"error": "task cannot be cancelled from state " + string(tk.State)}) return } - if err := s.store.UpdateTaskState(taskID, task.StateCancelled); err != nil { + if err := s.store.UpdateTaskStateBy(taskID, task.StateCancelled, event.ActorUser); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to cancel task"}) return } @@ -703,7 +704,7 @@ func (s *Server) handleAcceptTask(w http.ResponseWriter, r *http.Request) { }) return } - if err := s.store.UpdateTaskState(id, task.StateCompleted); err != nil { + if err := s.store.UpdateTaskStateBy(id, task.StateCompleted, event.ActorUser); err != nil { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } diff --git a/internal/storage/db.go b/internal/storage/db.go index 3a4f059..42d7800 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" ) @@ -261,8 +262,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 @@ -285,6 +295,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() } @@ -319,6 +337,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 } @@ -340,20 +367,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(¤tState); 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. diff --git a/internal/storage/events.go b/internal/storage/events.go index 82016ab..2051127 100644 --- a/internal/storage/events.go +++ b/internal/storage/events.go @@ -1,6 +1,7 @@ package storage import ( + "database/sql" "encoding/json" "errors" "fmt" @@ -8,12 +9,29 @@ import ( "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") } @@ -38,12 +56,6 @@ func (s *DB) CreateEvent(e *event.Event) error { 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) @@ -56,7 +68,16 @@ func (s *DB) CreateEvent(e *event.Event) error { ); err != nil { return fmt.Errorf("inserting event: %w", err) } - return tx.Commit() + 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 diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go index 3b342c7..1f03aeb 100644 --- a/internal/storage/events_test.go +++ b/internal/storage/events_test.go @@ -160,6 +160,112 @@ func TestListEvents_PreservesPayloadAndKind(t *testing.T) { } } +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 TestCreateEvent_RequiresFields(t *testing.T) { db := testDB(t) -- cgit v1.2.3