1 files changed, 51 insertions, 9 deletions

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(&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.