From 39cc108b5f0bf3d0238a1707d657a77b7a3df184 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
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/db.go | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

(limited to 'internal/storage/db.go')

diff --git a/internal/storage/db.go b/internal/storage/db.go
index 4adc1ba..3a4f059 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -139,6 +139,20 @@ 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`,
+		`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 {
-- 
cgit v1.2.3


From dc97d7ae00bd25668ba495fba249ceade0c6b100 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
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/db.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(&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.
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 <noreply@anthropic.com>
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/db.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 <noreply@anthropic.com>
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/db.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


From 39f74dbbc7adca0e2058112408bb99694deefcaf Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 26 May 2026 20:27:36 +0000
Subject: feat(budget,storage,config): per-provider spend accounting substrate
 (Phase 6)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds the budget accountant foundation:
- storage: a per-execution `agent` column (additive migration, populated by the
  dispatcher) and SpendByProviderSince(since), summing cost_usd per provider in
  a window. Accurate attribution survives a task running on different providers
  across retries.
- internal/budget: Accountant over a SpendSource + Limits, exposing Headroom
  (remaining/fraction per provider), Allow (would estCost breach the cap), and
  All (one query, sorted). Unconfigured/local providers are unlimited.
- config: a [budget] section (window + provider_5h_usd map). No default cap —
  gating is opt-in by configuring limits.

Fully unit-tested; dispatcher gating and the API/UI surface follow.

https://claude.ai/code/session_01SESwn7kQ7oP62trWw6pc39
---
 internal/budget/budget.go      | 122 +++++++++++++++++++++++++++++++++++++++++
 internal/budget/budget_test.go | 112 +++++++++++++++++++++++++++++++++++++
 internal/config/config.go      |  11 ++++
 internal/executor/executor.go  |   1 +
 internal/storage/db.go         |  50 +++++++++++++----
 internal/storage/db_test.go    |  38 +++++++++++++
 6 files changed, 324 insertions(+), 10 deletions(-)
 create mode 100644 internal/budget/budget.go
 create mode 100644 internal/budget/budget_test.go

(limited to 'internal/storage/db.go')

diff --git a/internal/budget/budget.go b/internal/budget/budget.go
new file mode 100644
index 0000000..60ce9d2
--- /dev/null
+++ b/internal/budget/budget.go
@@ -0,0 +1,122 @@
+// Package budget tracks agent spend against rolling per-provider windows so the
+// dispatcher can refuse (or reroute) paid work that would breach a cap. Local
+// runners are free and simply carry no configured limit.
+package budget
+
+import (
+	"sort"
+	"time"
+)
+
+// DefaultWindow is the rolling spend window when none is configured.
+const DefaultWindow = 5 * time.Hour
+
+// Limits configures per-provider spend caps within a rolling window.
+type Limits struct {
+	// Window is the rolling lookback; DefaultWindow when zero.
+	Window time.Duration
+	// PerProvider maps a provider (claude/gemini/…) to its USD cap within the
+	// window. A missing or non-positive entry means "unlimited" (e.g. local).
+	PerProvider map[string]float64
+}
+
+// SpendSource supplies total spend per provider since a given time.
+type SpendSource interface {
+	SpendByProviderSince(since time.Time) (map[string]float64, error)
+}
+
+// Headroom describes remaining budget for one provider in the current window.
+type Headroom struct {
+	Provider  string  `json:"provider"`
+	Limited   bool    `json:"limited"`
+	Limit     float64 `json:"limit_usd"`
+	Spent     float64 `json:"spent_usd"`
+	Remaining float64 `json:"remaining_usd"`
+	// Fraction is the share of the cap still available, 0..1 (0 when unlimited).
+	Fraction float64 `json:"fraction_remaining"`
+}
+
+// Accountant answers spend questions against a SpendSource and Limits.
+type Accountant struct {
+	src SpendSource
+	lim Limits
+	now func() time.Time
+}
+
+// New builds an Accountant. A zero Window defaults to DefaultWindow.
+func New(src SpendSource, lim Limits) *Accountant {
+	if lim.Window <= 0 {
+		lim.Window = DefaultWindow
+	}
+	return &Accountant{src: src, lim: lim, now: time.Now}
+}
+
+func (a *Accountant) since() time.Time { return a.now().Add(-a.lim.Window) }
+
+func (a *Accountant) headroomFrom(provider string, spends map[string]float64) Headroom {
+	limit := a.lim.PerProvider[provider]
+	if limit <= 0 {
+		return Headroom{Provider: provider, Limited: false}
+	}
+	spent := spends[provider]
+	remaining := limit - spent
+	if remaining < 0 {
+		remaining = 0
+	}
+	return Headroom{
+		Provider:  provider,
+		Limited:   true,
+		Limit:     limit,
+		Spent:     spent,
+		Remaining: remaining,
+		Fraction:  remaining / limit,
+	}
+}
+
+// Headroom returns the remaining budget for one provider in the current window.
+func (a *Accountant) Headroom(provider string) (Headroom, error) {
+	if a.lim.PerProvider[provider] <= 0 {
+		return Headroom{Provider: provider, Limited: false}, nil
+	}
+	spends, err := a.src.SpendByProviderSince(a.since())
+	if err != nil {
+		return Headroom{}, err
+	}
+	return a.headroomFrom(provider, spends), nil
+}
+
+// Allow reports whether starting a task on provider that may cost up to estCost
+// is permitted without breaching the window cap. Unlimited providers always
+// pass. estCost is typically the task's max_budget_usd (0 = unknown).
+func (a *Accountant) Allow(provider string, estCost float64) (bool, error) {
+	h, err := a.Headroom(provider)
+	if err != nil {
+		return false, err
+	}
+	if !h.Limited {
+		return true, nil
+	}
+	return h.Spent+estCost <= h.Limit, nil
+}
+
+// All returns headroom for every configured provider, sorted by name, in a
+// single spend query.
+func (a *Accountant) All() ([]Headroom, error) {
+	if len(a.lim.PerProvider) == 0 {
+		return []Headroom{}, nil
+	}
+	spends, err := a.src.SpendByProviderSince(a.since())
+	if err != nil {
+		return nil, err
+	}
+	providers := make([]string, 0, len(a.lim.PerProvider))
+	for p := range a.lim.PerProvider {
+		providers = append(providers, p)
+	}
+	sort.Strings(providers)
+	out := make([]Headroom, 0, len(providers))
+	for _, p := range providers {
+		out = append(out, a.headroomFrom(p, spends))
+	}
+	return out, nil
+}
diff --git a/internal/budget/budget_test.go b/internal/budget/budget_test.go
new file mode 100644
index 0000000..c257c91
--- /dev/null
+++ b/internal/budget/budget_test.go
@@ -0,0 +1,112 @@
+package budget
+
+import (
+	"testing"
+	"time"
+)
+
+type fakeSpend struct {
+	spends map[string]float64
+	since  time.Time
+}
+
+func (f *fakeSpend) SpendByProviderSince(since time.Time) (map[string]float64, error) {
+	f.since = since
+	return f.spends, nil
+}
+
+func newAcct(spends map[string]float64, lim Limits) (*Accountant, *fakeSpend) {
+	src := &fakeSpend{spends: spends}
+	a := New(src, lim)
+	a.now = func() time.Time { return time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC) }
+	return a, src
+}
+
+func TestNew_DefaultsWindow(t *testing.T) {
+	a, _ := newAcct(nil, Limits{})
+	if a.lim.Window != DefaultWindow {
+		t.Errorf("want default window %v, got %v", DefaultWindow, a.lim.Window)
+	}
+}
+
+func TestHeadroom_UnlimitedWhenNoLimit(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"local": 999}, Limits{PerProvider: map[string]float64{"claude": 10}})
+	h, err := a.Headroom("local")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if h.Limited {
+		t.Errorf("local should be unlimited, got %+v", h)
+	}
+}
+
+func TestHeadroom_ComputesRemainingAndFraction(t *testing.T) {
+	a, src := newAcct(map[string]float64{"claude": 4}, Limits{Window: 5 * time.Hour, PerProvider: map[string]float64{"claude": 10}})
+	h, err := a.Headroom("claude")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !h.Limited || h.Limit != 10 || h.Spent != 4 || h.Remaining != 6 {
+		t.Errorf("unexpected headroom: %+v", h)
+	}
+	if h.Fraction != 0.6 {
+		t.Errorf("fraction: want 0.6, got %v", h.Fraction)
+	}
+	// Window lookback is now-5h.
+	want := time.Date(2026, 5, 26, 7, 0, 0, 0, time.UTC)
+	if !src.since.Equal(want) {
+		t.Errorf("since: want %v, got %v", want, src.since)
+	}
+}
+
+func TestHeadroom_ClampsNegativeRemaining(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"claude": 15}, Limits{PerProvider: map[string]float64{"claude": 10}})
+	h, _ := a.Headroom("claude")
+	if h.Remaining != 0 {
+		t.Errorf("remaining should clamp to 0, got %v", h.Remaining)
+	}
+}
+
+func TestAllow(t *testing.T) {
+	a, _ := newAcct(map[string]float64{"claude": 8}, Limits{PerProvider: map[string]float64{"claude": 10}})
+
+	if ok, _ := a.Allow("claude", 1.5); !ok {
+		t.Error("8 + 1.5 <= 10 should be allowed")
+	}
+	if ok, _ := a.Allow("claude", 3); ok {
+		t.Error("8 + 3 > 10 should be denied")
+	}
+	if ok, _ := a.Allow("local", 1000); !ok {
+		t.Error("unlimited provider should always allow")
+	}
+}
+
+func TestAll_SortedAndSingleQuery(t *testing.T) {
+	a, src := newAcct(map[string]float64{"claude": 2, "gemini": 1},
+		Limits{PerProvider: map[string]float64{"gemini": 5, "claude": 10}})
+	calls := 0
+	orig := src.spends
+	a2 := New(&countingSpend{spends: orig, calls: &calls}, a.lim)
+	a2.now = a.now
+
+	hs, err := a2.All()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(hs) != 2 || hs[0].Provider != "claude" || hs[1].Provider != "gemini" {
+		t.Errorf("want [claude gemini] sorted, got %+v", hs)
+	}
+	if calls != 1 {
+		t.Errorf("All should issue exactly one spend query, got %d", calls)
+	}
+}
+
+type countingSpend struct {
+	spends map[string]float64
+	calls  *int
+}
+
+func (c *countingSpend) SpendByProviderSince(time.Time) (map[string]float64, error) {
+	*c.calls++
+	return c.spends, nil
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index 94f3ec7..58de95c 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -69,6 +69,17 @@ type Config struct {
 	VAPIDEmail       string        `toml:"vapid_email"`
 	ClaudeConfigDir  string        `toml:"claude_config_dir"`
 	LocalModel       LocalModel    `toml:"local_model"`
+	Budget           BudgetConfig  `toml:"budget"`
+}
+
+// BudgetConfig configures rolling per-provider spend caps. With no providers
+// set, budget gating is disabled (nothing is blocked) — there is intentionally
+// no default cap; the user must opt in by configuring limits.
+type BudgetConfig struct {
+	// Window is the rolling lookback duration (e.g. "5h"); defaults to 5h.
+	Window string `toml:"window"`
+	// Provider5hUSD maps a provider (claude/gemini) to its USD cap within Window.
+	Provider5hUSD map[string]float64 `toml:"provider_5h_usd"`
 }
 
 func Default() (*Config, error) {
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 6d6c528..4333f51 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -1025,6 +1025,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
 		TaskID:    t.ID,
 		StartTime: time.Now().UTC(),
 		Status:    "RUNNING",
+		Agent:     agentType,
 	}
 
 	// Pre-populate log paths so they're available in the DB immediately —
diff --git a/internal/storage/db.go b/internal/storage/db.go
index adb7c02..e03a902 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -140,6 +140,7 @@ 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,
@@ -536,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
@@ -584,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
 	}
@@ -601,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
@@ -621,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
 	}
@@ -656,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)
 }
 
@@ -1137,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
 	}
@@ -1146,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)
-- 
cgit v1.2.3