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

1 files changed, 56 insertions, 7 deletions

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()
 }