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>

1 files changed, 190 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 {