From 39cc108b5f0bf3d0238a1707d657a77b7a3df184 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 09:08:52 +0000 Subject: feat(event,storage): add events table substrate for OSS redesign New internal/event package defines the Event type with canonical Kind and Actor constants. New events table is appended via additive migration (id, task_id, seq, ts, kind, actor, payload_json) with seq as per-task monotonic, assigned by the storage layer inside a transaction. CreateEvent + ListEvents support the future migration of question_json, interactions_json, summary, and elaboration_input off the tasks table into a unified, replayable event stream. No existing code paths are modified yet; this is pure substrate for Phase 1. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/storage/events_test.go | 182 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 internal/storage/events_test.go (limited to 'internal/storage/events_test.go') 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") + } + }) + } +} -- cgit v1.2.3 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/storage/events_test.go') 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 From b72978f4454235a00a2e5c92b5332b985cc08fa9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 23 May 2026 08:10:03 +0000 Subject: feat(storage): dual-write question/summary/interaction flows to events The legacy task columns (question_json, summary, interactions_json) now also emit events in the same transaction as the column update: - UpdateTaskQuestion(non-empty) -> clarification_request (agent); the empty-string clear-on-answer emits nothing - UpdateTaskSummary -> summary (agent) - AppendTaskInteraction -> clarification_answer (user) This populates the event stream from the existing Q&A and summary paths without removing the old columns yet, setting up their eventual deletion in the cleanup phase. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/storage/db.go | 63 +++++++++++++++++++++--- internal/storage/events_test.go | 103 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 159 insertions(+), 7 deletions(-) (limited to 'internal/storage/events_test.go') diff --git a/internal/storage/db.go b/internal/storage/db.go index 42d7800..a871c71 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -880,17 +880,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. @@ -943,6 +983,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() } diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go index 1f03aeb..3ce79ab 100644 --- a/internal/storage/events_test.go +++ b/internal/storage/events_test.go @@ -266,6 +266,109 @@ func TestRejectTask_WritesUserStateChangeEvent(t *testing.T) { } } +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 TestCreateEvent_RequiresFields(t *testing.T) { db := testDB(t) -- cgit v1.2.3 From 48a8e49cdda833f8af22e839ca3c102c40c97e17 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 24 May 2026 08:19:23 +0000 Subject: feat(storage): emit subtask_spawned events on the parent stream CreateTask is now transactional and, when a task has a parent_task_id, records a subtask_spawned event (actor agent) on the parent's stream so the parent's history shows the children it dispatched. Top-level task creation (no parent) emits nothing. This closes the one agent-originated signal that flows during a run today (subtasks POSTed to the API per the planning preamble) but previously left no trace in the event stream. https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39 --- internal/storage/db.go | 32 ++++++++++++++++++-- internal/storage/events_test.go | 65 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 3 deletions(-) (limited to 'internal/storage/events_test.go') diff --git a/internal/storage/db.go b/internal/storage/db.go index a871c71..adb7c02 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -190,15 +190,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. diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go index 3ce79ab..c5d3c37 100644 --- a/internal/storage/events_test.go +++ b/internal/storage/events_test.go @@ -369,6 +369,71 @@ func TestAppendTaskInteraction_EmitsClarificationAnswer(t *testing.T) { } } +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) -- cgit v1.2.3