diff options
Diffstat (limited to 'internal/storage/db.go')
| -rw-r--r-- | internal/storage/db.go | 36 |
1 files changed, 30 insertions, 6 deletions
diff --git a/internal/storage/db.go b/internal/storage/db.go index 8b563a6..51ea648 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -184,6 +184,11 @@ func (s *DB) migrate() error { `CREATE INDEX IF NOT EXISTS idx_stories_status ON stories(status)`, `CREATE INDEX IF NOT EXISTS idx_stories_epic_id ON stories(epic_id)`, `CREATE INDEX IF NOT EXISTS idx_stories_root_task_id ON stories(root_task_id)`, + // needs_review (Phase 7c): flagged by internal/scheduler.Scheduler + // when it resumes a role-typed task past its ask_user-timeout with a + // system-authored fallback answer, so a human can find and double-check + // it later via GET /api/tasks?needs_review=true. + `ALTER TABLE tasks ADD COLUMN needs_review BOOLEAN NOT NULL DEFAULT 0`, } for _, m := range migrations { if _, err := s.db.Exec(m); err != nil { @@ -258,13 +263,13 @@ func (s *DB) CreateTask(t *task.Task) error { // GetTask retrieves a task by ID. func (s *DB) GetTask(id string) (*task.Task, error) { - row := s.db.QueryRow(`SELECT 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, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE id = ?`, id) + row := s.db.QueryRow(`SELECT 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, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE id = ?`, id) return scanTask(row) } // ListTasks returns tasks matching the given filter. func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) { - query := `SELECT 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, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE 1=1` + query := `SELECT 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, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE 1=1` var args []interface{} if filter.State != "" { @@ -275,6 +280,10 @@ func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) { query += " AND updated_at > ?" args = append(args, filter.Since.UTC()) } + if filter.NeedsReview != nil { + query += " AND needs_review = ?" + args = append(args, *filter.NeedsReview) + } query += " ORDER BY created_at DESC" if filter.Limit > 0 { query += " LIMIT ?" @@ -300,7 +309,7 @@ func (s *DB) ListTasks(filter TaskFilter) ([]*task.Task, error) { // ListSubtasks returns all tasks whose parent_task_id matches the given ID. func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) { - rows, err := s.db.Query(`SELECT 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, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE parent_task_id = ? ORDER BY created_at ASC`, parentID) + rows, err := s.db.Query(`SELECT 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, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE parent_task_id = ? ORDER BY created_at ASC`, parentID) if err != nil { return nil, err } @@ -325,7 +334,7 @@ func (s *DB) ListSubtasks(parentID string) ([]*task.Task, error) { // Only direct dependents are returned — callers that need the full // transitive downstream subtree (e.g. cascade-cancellation) must recurse. func (s *DB) ListDependents(taskID string) ([]*task.Task, error) { - rows, err := s.db.Query(`SELECT 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, rejection_comment, question_json, summary, interactions_json FROM tasks`) + rows, err := s.db.Query(`SELECT 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, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks`) if err != nil { return nil, err } @@ -400,7 +409,7 @@ func (s *DB) ResetTaskForRetry(id string) (*task.Task, error) { } defer tx.Rollback() //nolint:errcheck - t, err := scanTask(tx.QueryRow(`SELECT 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, rejection_comment, question_json, summary, interactions_json FROM tasks WHERE id = ?`, id)) + t, err := scanTask(tx.QueryRow(`SELECT 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, rejection_comment, question_json, summary, interactions_json, needs_review FROM tasks WHERE id = ?`, id)) if err != nil { if err == sql.ErrNoRows { return nil, fmt.Errorf("task %q not found", id) @@ -450,6 +459,18 @@ func (s *DB) UpdateTaskAgent(id string, agent task.AgentConfig) error { return err } +// UpdateTaskNeedsReview sets tasks.needs_review. Used by +// internal/scheduler.Scheduler to flag a task whose BLOCKED ask_user question +// was resumed with a system-authored fallback answer after timing out, +// rather than a real human answer, so a human can find it later via +// GET /api/tasks?needs_review=true. +func (s *DB) UpdateTaskNeedsReview(id string, needsReview bool) error { + now := time.Now().UTC() + _, err := s.db.Exec(`UPDATE tasks SET needs_review = ?, updated_at = ? WHERE id = ?`, + needsReview, now, id) + return err +} + // RejectTask sets a task's state to PENDING and stores the rejection comment. func (s *DB) RejectTask(id, comment string) error { tx, err := s.db.Begin() @@ -550,6 +571,9 @@ type TaskFilter struct { State task.State Limit int Since time.Time + // NeedsReview, when non-nil, filters to tasks.needs_review == *NeedsReview. + // nil (the default) applies no filter. Backs GET /api/tasks?needs_review=true. + NeedsReview *bool } // GetMaxUpdatedAt returns the most recent updated_at timestamp across all tasks. @@ -1156,7 +1180,7 @@ func scanTask(row scanner) (*task.Task, error) { &t.ID, &t.Name, &t.Description, &elaborationInput, &project, &repositoryURL, &configJSON, &priority, &timeoutNS, &retryJSON, &tagsJSON, &depsJSON, &parentTaskID, &state, &t.CreatedAt, &t.UpdatedAt, - &rejectionComment, &questionJSON, &summary, &interactionsJSON, + &rejectionComment, &questionJSON, &summary, &interactionsJSON, &t.NeedsReview, ) t.ParentTaskID = parentTaskID.String t.ElaborationInput = elaborationInput.String |