nova28
8898459c0b
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
2907 lines
104 KiB
Go
2907 lines
104 KiB
Go
package sqlite
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jmoiron/sqlx"
|
|
|
|
"github.com/kandev/kandev/internal/agentctl/tracing"
|
|
internaldb "github.com/kandev/kandev/internal/db"
|
|
"github.com/kandev/kandev/internal/db/dialect"
|
|
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
|
|
"github.com/kandev/kandev/internal/steptelemetry"
|
|
"github.com/kandev/kandev/internal/task/models"
|
|
usermodels "github.com/kandev/kandev/internal/user/models"
|
|
wfmodels "github.com/kandev/kandev/internal/workflow/models"
|
|
v1 "github.com/kandev/kandev/pkg/api/v1"
|
|
)
|
|
|
|
// defaultTaskAlias is the fallback alias the projection helpers use when
|
|
// the caller passes an empty string — i.e., when the SELECT references
|
|
// the `tasks` table directly rather than through a join alias.
|
|
const defaultTaskAlias = "tasks"
|
|
|
|
type taskScanColumn struct {
|
|
name string
|
|
selectExpr func(alias string) string
|
|
}
|
|
|
|
// Automation runs are hidden from the board and from task lists by their
|
|
// provenance, not by ephemerality: their tasks are ordinary persistent tasks
|
|
// that keep their worktree and stay repliable, they just have their own
|
|
// destination (docs/specs/office/automations-settings.md). is_ephemeral keeps
|
|
// its original quick-chat meaning, so every board read pairs the two.
|
|
const (
|
|
andNotAutomationOrigin = ` AND COALESCE(origin, '') != '` + models.TaskOriginAutomationRun + `'`
|
|
andNotAutomationOriginT = ` AND COALESCE(t.origin, '') != '` + models.TaskOriginAutomationRun + `'`
|
|
)
|
|
|
|
var taskScanColumns = []taskScanColumn{
|
|
{name: "id"},
|
|
{name: "workspace_id"},
|
|
{name: "workflow_id"},
|
|
{name: "workflow_step_id"},
|
|
{name: "title"},
|
|
{name: "description"},
|
|
{name: "state"},
|
|
{name: "priority"},
|
|
{name: "position"},
|
|
{name: "wip_admitted"},
|
|
{name: "queued_for_step_id"},
|
|
{name: "queued_at"},
|
|
{name: "metadata"},
|
|
{name: "is_ephemeral"},
|
|
{name: "parent_id"},
|
|
{name: "autopilot_enabled"},
|
|
{name: "archived_at"},
|
|
{name: "archived_by_cascade_id", selectExpr: func(alias string) string {
|
|
return `COALESCE(` + alias + `.archived_by_cascade_id, '') AS archived_by_cascade_id`
|
|
}},
|
|
{name: "created_at"},
|
|
{name: "updated_at"},
|
|
{name: "assignee_agent_profile_id", selectExpr: func(alias string) string {
|
|
return runnerProjection(alias) + ` AS assignee_agent_profile_id`
|
|
}},
|
|
{name: "origin"},
|
|
{name: "project_id"},
|
|
{name: "labels"},
|
|
{name: "identifier"},
|
|
{name: "external_id"},
|
|
{name: "external_id_settled_at"},
|
|
{name: "is_from_office", selectExpr: func(alias string) string {
|
|
return isFromOfficeProjection(alias) + ` AS is_from_office`
|
|
}},
|
|
}
|
|
|
|
// taskSelectColumns returns the column projection (with runner subquery)
|
|
// for a SELECT against tasks aliased as `alias`. The output column order
|
|
// matches scanSingleTask / scanTasks.
|
|
func taskSelectColumns(alias string) string {
|
|
if alias == "" {
|
|
alias = defaultTaskAlias
|
|
}
|
|
return taskScanColumnSQL(alias, true)
|
|
}
|
|
|
|
func taskProjectedColumns(alias string) string {
|
|
if alias == "" {
|
|
alias = defaultTaskAlias
|
|
}
|
|
return taskScanColumnSQL(alias, false)
|
|
}
|
|
|
|
func taskScanColumnSQL(alias string, useExpressions bool) string {
|
|
prefix := alias + "."
|
|
cols := make([]string, 0, len(taskScanColumns))
|
|
for _, col := range taskScanColumns {
|
|
if useExpressions && col.selectExpr != nil {
|
|
cols = append(cols, col.selectExpr(alias))
|
|
continue
|
|
}
|
|
cols = append(cols, prefix+col.name)
|
|
}
|
|
return strings.Join(cols, ", ")
|
|
}
|
|
|
|
// isFromOfficeProjection returns a SQL boolean expression that is true
|
|
// when the task is owned by office: either it has a non-empty project_id
|
|
// (explicit office task) or its workflow matches the workspace's
|
|
// office_workflow_id (the canonical "office workflow"). Kanban tasks live
|
|
// in any other workflow and have no project.
|
|
func isFromOfficeProjection(alias string) string {
|
|
if alias == "" {
|
|
alias = defaultTaskAlias
|
|
}
|
|
return `(
|
|
COALESCE(` + alias + `.project_id, '') != ''
|
|
OR EXISTS (
|
|
SELECT 1 FROM workspaces w
|
|
WHERE w.id = ` + alias + `.workspace_id
|
|
AND COALESCE(w.office_workflow_id, '') != ''
|
|
AND w.office_workflow_id = ` + alias + `.workflow_id
|
|
)
|
|
)`
|
|
}
|
|
|
|
// IsFromOfficePredicate returns the shared SQL predicate for authoritative
|
|
// Office-task identity. Callers that query tasks outside this repository must
|
|
// use this expression so project-linked tasks and canonical Office-workflow
|
|
// tasks are classified consistently with models.Task.IsFromOffice.
|
|
func IsFromOfficePredicate(alias string) string {
|
|
return isFromOfficeProjection(alias)
|
|
}
|
|
|
|
// excludeConfigModePredicate delegates to the shared dialect helper (also
|
|
// used by internal/analytics/repository/sqlite's CodeStats read, so both
|
|
// cover the same office config-mode exclusion).
|
|
func excludeConfigModePredicate(driver, col string) string {
|
|
return dialect.ExcludeConfigModePredicate(driver, col)
|
|
}
|
|
|
|
// runnerProjection produces the correlated subquery (without alias) that
|
|
// resolves the runner for the row of `tasks` with the given alias. Used
|
|
// inline in SELECT projections.
|
|
func runnerProjection(alias string) string {
|
|
if alias == "" {
|
|
alias = defaultTaskAlias
|
|
}
|
|
return `COALESCE(
|
|
(SELECT wsp.agent_profile_id FROM workflow_step_participants wsp
|
|
WHERE wsp.step_id = ` + alias + `.workflow_step_id
|
|
AND wsp.task_id = ` + alias + `.id
|
|
AND wsp.role = 'runner'
|
|
ORDER BY wsp.position ASC, wsp.id ASC LIMIT 1),
|
|
(SELECT ws.agent_profile_id FROM workflow_steps ws WHERE ws.id = ` + alias + `.workflow_step_id),
|
|
''
|
|
)`
|
|
}
|
|
|
|
// CreateTask creates a new task. The assignee column has been removed
|
|
// (ADR 0005 Wave F); when the request carries AssigneeAgentProfileID we
|
|
// upsert a 'runner' row in workflow_step_participants instead.
|
|
func (r *Repository) CreateTask(ctx context.Context, task *models.Task) error {
|
|
if task.WorkflowStepID != "" && task.QueuedForStepID == "" && !task.IsEphemeral {
|
|
task.WIPAdmitted = true
|
|
models.DropWIPDeferredLaunch(task)
|
|
}
|
|
return r.createTask(ctx, task, "", 0)
|
|
}
|
|
|
|
// CreateTaskIfWorkflowStepHasCapacity atomically admits a task into a
|
|
// WIP-limited workflow step. The occupancy check and insert share one writer
|
|
// transaction, so concurrent watcher events cannot overfill the step.
|
|
func (r *Repository) CreateTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, targetStepID string, limit int) error {
|
|
if limit <= 0 {
|
|
return r.CreateTask(ctx, task)
|
|
}
|
|
task.WIPAdmitted = true
|
|
task.QueuedForStepID = ""
|
|
task.QueuedAt = nil
|
|
return r.createTask(ctx, task, targetStepID, limit)
|
|
}
|
|
|
|
// CreateTaskWithWorkflowStepAdmission persists a task according to the
|
|
// destination step's WIP state. Overflow is either kept visibly queued in the
|
|
// destination or placed in the configured feeder for one-hop promotion.
|
|
func (r *Repository) CreateTaskWithWorkflowStepAdmission(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
targetStepID string,
|
|
targetLimit int,
|
|
feederStepID string,
|
|
feederLimit int,
|
|
) error {
|
|
if task.IsEphemeral || targetStepID == "" || targetLimit <= 0 {
|
|
task.WIPAdmitted = !task.IsEphemeral
|
|
task.QueuedForStepID = ""
|
|
task.QueuedAt = nil
|
|
models.DropWIPDeferredLaunch(task)
|
|
return r.CreateTask(ctx, task)
|
|
}
|
|
|
|
if err := r.prepareTaskForCreate(task); err != nil {
|
|
return err
|
|
}
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
// Same order as createTask and the workspace cascade: workspace row
|
|
// first, then workflow-step locks. Without it an admission holding a step
|
|
// lock while the cascade holds the workspace and waits for that step
|
|
// deadlocks on Postgres.
|
|
if err := r.lockWorkspaceRowStdTx(ctx, tx, task.WorkspaceID); err != nil {
|
|
return err
|
|
}
|
|
if err := r.lockWorkflowStepsForAdmission(ctx, tx, targetStepID, feederStepID); err != nil {
|
|
return err
|
|
}
|
|
targetOccupants, err := r.countAdmittedInTx(ctx, tx, targetStepID, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := r.applyAdmissionPlacement(ctx, tx, task, targetStepID, targetLimit, feederStepID, feederLimit, targetOccupants); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := r.insertTaskTx(ctx, tx, task); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (r *Repository) applyAdmissionPlacement(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
task *models.Task,
|
|
targetStepID string,
|
|
targetLimit int,
|
|
feederStepID string,
|
|
feederLimit int,
|
|
targetOccupants int,
|
|
) error {
|
|
switch {
|
|
case targetOccupants < targetLimit:
|
|
task.WorkflowStepID = targetStepID
|
|
task.WIPAdmitted = true
|
|
task.QueuedForStepID = ""
|
|
task.QueuedAt = nil
|
|
models.DropWIPDeferredLaunch(task)
|
|
case feederStepID == "":
|
|
task.WorkflowStepID = targetStepID
|
|
task.WIPAdmitted = false
|
|
task.QueuedForStepID = targetStepID
|
|
task.QueuedAt = &task.CreatedAt
|
|
case feederStepID == targetStepID:
|
|
return wfmodels.NewWIPLimitError(targetStepID, targetLimit, targetOccupants)
|
|
default:
|
|
feederOccupants, err := r.countAdmittedInTx(ctx, tx, feederStepID, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if feederLimit > 0 && feederOccupants >= feederLimit {
|
|
return wfmodels.NewWIPLimitError(feederStepID, feederLimit, feederOccupants)
|
|
}
|
|
task.WorkflowStepID = feederStepID
|
|
task.WIPAdmitted = true
|
|
task.QueuedForStepID = targetStepID
|
|
task.QueuedAt = &task.CreatedAt
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) createTask(ctx context.Context, task *models.Task, targetStepID string, limit int) error {
|
|
if err := r.prepareTaskForCreate(task); err != nil {
|
|
return err
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
// Serialize with the workspace delete cascade: the cascade locks the
|
|
// workspace row before inventorying its tasks, so a task created here
|
|
// either commits before the cascade's inventory (and is purged with the
|
|
// rest) or blocks until the cascade finishes, when the workspace is gone
|
|
// and the insert fails its foreign key. The workspace lock is taken
|
|
// before any workflow-step lock so the creation/admission paths share one
|
|
// order with the cascade.
|
|
if err := r.lockWorkspaceRowStdTx(ctx, tx, task.WorkspaceID); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := r.ensureWorkflowStepCapacity(ctx, tx, targetStepID, limit); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := r.insertTaskTx(ctx, tx, task); err != nil {
|
|
if rollbackErr := tx.Rollback(); rollbackErr != nil {
|
|
return fmt.Errorf("failed to rollback task insert: %w", rollbackErr)
|
|
}
|
|
return err
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (r *Repository) prepareTaskForCreate(task *models.Task) error {
|
|
if task.ID == "" {
|
|
task.ID = uuid.New().String()
|
|
}
|
|
now := time.Now().UTC()
|
|
task.CreatedAt = now
|
|
task.UpdatedAt = now
|
|
if task.Metadata == nil {
|
|
task.Metadata = map[string]interface{}{}
|
|
}
|
|
if task.Labels == "" {
|
|
task.Labels = "[]"
|
|
}
|
|
if task.Priority == "" {
|
|
task.Priority = "medium"
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) insertTaskTx(ctx context.Context, tx *sql.Tx, task *models.Task) error {
|
|
metadata, err := json.Marshal(task.Metadata)
|
|
if err != nil {
|
|
metadata = []byte("{}")
|
|
}
|
|
var externalID interface{}
|
|
if task.ExternalID != "" {
|
|
externalID = task.ExternalID
|
|
}
|
|
_, err = tx.ExecContext(ctx, r.db.Rebind(`
|
|
INSERT INTO tasks (id, workspace_id, workflow_id, workflow_step_id, title, description, state, priority, position, wip_admitted, queued_for_step_id, queued_at, metadata, is_ephemeral, parent_id, autopilot_enabled, created_at, updated_at, origin, project_id, labels, identifier, external_id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`), task.ID, task.WorkspaceID, task.WorkflowID, task.WorkflowStepID, task.Title, task.Description, task.State, task.Priority, task.Position, dialect.BoolToInt(task.WIPAdmitted), task.QueuedForStepID, task.QueuedAt, string(metadata), dialect.BoolToInt(task.IsEphemeral), task.ParentID, dialect.BoolToInt(task.Autopilot), task.CreatedAt, task.UpdatedAt, task.Origin, task.ProjectID, task.Labels, task.Identifier, externalID)
|
|
if err != nil {
|
|
if isExternalIDUniqueViolation(err) {
|
|
return fmt.Errorf("%w: %w", ErrExternalIDConflict, err)
|
|
}
|
|
return err
|
|
}
|
|
// Genesis ledger row. By this point applyAdmissionPlacement has already
|
|
// rewritten task.WorkflowStepID to the actual placement (feeder step when
|
|
// WIP diverted it), so this satisfies the spec's feeder-step scenario for
|
|
// free. A task created with no workflow writes nothing.
|
|
genesisCtx := steptelemetry.WithAttribution(ctx, genesisAttribution(ctx))
|
|
if err := r.recordStepTransition(genesisCtx, tx, stepTransitionInput{
|
|
taskID: task.ID,
|
|
toWorkflowID: task.WorkflowID,
|
|
toWorkflowStepID: task.WorkflowStepID,
|
|
occurredAt: task.CreatedAt,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if task.AssigneeAgentProfileID != "" && task.WorkflowStepID != "" {
|
|
return upsertRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) lockWorkflowStepsForAdmission(ctx context.Context, tx *sql.Tx, targetStepID, feederStepID string) error {
|
|
ids := []string{targetStepID}
|
|
if feederStepID != "" && feederStepID != targetStepID {
|
|
ids = append(ids, feederStepID)
|
|
}
|
|
if ids[0] > ids[len(ids)-1] && len(ids) == 2 {
|
|
ids[0], ids[1] = ids[1], ids[0]
|
|
}
|
|
for _, id := range ids {
|
|
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, id); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) countAdmittedInTx(ctx context.Context, tx *sql.Tx, stepID, excludeTaskID string) (int, error) {
|
|
query := `SELECT COUNT(*) FROM tasks WHERE workflow_step_id = ? AND wip_admitted = 1 AND archived_at IS NULL AND is_ephemeral = 0` + andNotAutomationOrigin
|
|
args := []interface{}{stepID}
|
|
if excludeTaskID != "" {
|
|
query += " AND id != ?"
|
|
args = append(args, excludeTaskID)
|
|
}
|
|
var count int
|
|
err := tx.QueryRowContext(ctx, r.db.Rebind(query), args...).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
func (r *Repository) ensureWorkflowStepCapacity(ctx context.Context, tx *sql.Tx, targetStepID string, limit int) error {
|
|
if limit <= 0 {
|
|
return nil
|
|
}
|
|
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, targetStepID); err != nil {
|
|
return err
|
|
}
|
|
var occupants int
|
|
if err := tx.QueryRowContext(ctx, r.db.Rebind(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE workflow_step_id = ?
|
|
AND wip_admitted = 1
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), targetStepID).Scan(&occupants); err != nil {
|
|
return err
|
|
}
|
|
if occupants >= limit {
|
|
return wfmodels.NewWIPLimitError(targetStepID, limit, occupants)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func lockWorkflowStepForCapacity(ctx context.Context, tx *sql.Tx, driver string, rebind func(string) string, stepID string) error {
|
|
if !dialect.IsPostgres(driver) {
|
|
return nil
|
|
}
|
|
var lockedID string
|
|
return tx.QueryRowContext(ctx, rebind(`SELECT id FROM workflow_steps WHERE id = ? FOR UPDATE`), stepID).Scan(&lockedID)
|
|
}
|
|
|
|
// upsertRunnerInTx writes (or replaces) a 'runner' participant row for
|
|
// (stepID, taskID) inside the provided transaction. Mirrors
|
|
// workflow.Repository.SetTaskRunner but reuses the caller's tx.
|
|
//
|
|
// rebind is the caller's r.db.Rebind — required on Postgres, where the raw
|
|
// "?" placeholders below are not valid bind syntax (unlike SQLite, which
|
|
// accepts them natively). Mirrors lockWorkflowStepForCapacity's pattern for
|
|
// a free function that isn't a *Repository method.
|
|
func upsertRunnerInTx(ctx context.Context, tx *sql.Tx, rebind func(string) string, stepID, taskID, agentProfileID string) error {
|
|
if stepID == "" || taskID == "" || agentProfileID == "" {
|
|
return nil
|
|
}
|
|
var existing string
|
|
err := tx.QueryRowContext(ctx, rebind(`SELECT id FROM workflow_step_participants
|
|
WHERE step_id = ? AND task_id = ? AND role = 'runner' LIMIT 1`),
|
|
stepID, taskID).Scan(&existing)
|
|
if err == nil {
|
|
_, uerr := tx.ExecContext(ctx,
|
|
rebind(`UPDATE workflow_step_participants SET agent_profile_id = ? WHERE id = ?`),
|
|
agentProfileID, existing)
|
|
return uerr
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
id := uuid.New().String()
|
|
_, ierr := tx.ExecContext(ctx, rebind(`INSERT INTO workflow_step_participants
|
|
(id, step_id, task_id, role, agent_profile_id, decision_required, position)
|
|
VALUES (?, ?, ?, 'runner', ?, 0, 0)`),
|
|
id, stepID, taskID, agentProfileID)
|
|
return ierr
|
|
}
|
|
|
|
// clearRunnerInTx removes any 'runner' participant row for (stepID, taskID).
|
|
func clearRunnerInTx(ctx context.Context, tx *sql.Tx, rebind func(string) string, stepID, taskID string) error {
|
|
if stepID == "" || taskID == "" {
|
|
return nil
|
|
}
|
|
_, err := tx.ExecContext(ctx,
|
|
rebind(`DELETE FROM workflow_step_participants
|
|
WHERE step_id = ? AND task_id = ? AND role = 'runner'`),
|
|
stepID, taskID)
|
|
return err
|
|
}
|
|
|
|
// syncRunnerInTx upserts the runner participant when agentProfileID is set,
|
|
// otherwise clears it. No-op when stepID is empty.
|
|
func syncRunnerInTx(ctx context.Context, tx *sql.Tx, rebind func(string) string, stepID, taskID, agentProfileID string) error {
|
|
if stepID == "" {
|
|
return nil
|
|
}
|
|
if agentProfileID != "" {
|
|
return upsertRunnerInTx(ctx, tx, rebind, stepID, taskID, agentProfileID)
|
|
}
|
|
return clearRunnerInTx(ctx, tx, rebind, stepID, taskID)
|
|
}
|
|
|
|
// GetTask retrieves a task by ID
|
|
func (r *Repository) GetTask(ctx context.Context, id string) (*models.Task, error) {
|
|
row := r.ro.QueryRowContext(ctx, r.ro.Rebind(
|
|
`SELECT `+taskSelectColumns("t")+` FROM tasks t WHERE t.id = ?`), id)
|
|
task, err := r.scanSingleTask(row)
|
|
if err == sql.ErrNoRows {
|
|
return nil, fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
return task, err
|
|
}
|
|
|
|
// UpdateTask updates an existing task. The runner write lands as an
|
|
// upsert/clear on workflow_step_participants inside the same tx as the
|
|
// task UPDATE.
|
|
func (r *Repository) UpdateTask(ctx context.Context, task *models.Task) error {
|
|
metadata, err := json.Marshal(task.Metadata)
|
|
if err != nil {
|
|
metadata = []byte("{}")
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
if err := r.updateTaskTx(ctx, tx, task, metadata); err != nil {
|
|
return err
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (r *Repository) updateTaskTx(ctx context.Context, tx *sql.Tx, task *models.Task, metadata []byte) error {
|
|
fromWorkflowID, fromStepID, _, err := r.readTaskStepInTx(ctx, tx, task.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Stamped after the transactional read/lock above, not before BeginTx: on
|
|
// Postgres, readTaskStepInTx's FOR UPDATE blocks until this transaction's
|
|
// turn to touch the row, so the timestamp now reflects true serialization
|
|
// order. Stamping it earlier let two concurrent movers commit out of
|
|
// timestamp order relative to their actual commit order, which broke the
|
|
// (occurred_at, id) chain invariant under real concurrent load — SQLite's
|
|
// single-writer connection pool serializes callers regardless, so this
|
|
// was invisible until exercised against Postgres with real concurrency.
|
|
task.UpdatedAt = r.nowUTC()
|
|
|
|
updateQuery := `
|
|
UPDATE tasks SET workspace_id = ?, workflow_id = ?, workflow_step_id = ?, title = ?, description = ?, state = ?, priority = ?, position = ?, wip_admitted = ?, queued_for_step_id = ?, queued_at = ?, metadata = ?, parent_id = ?, updated_at = ?, origin = ?, project_id = ?, labels = ?, identifier = ?
|
|
WHERE id = ?
|
|
`
|
|
if models.IsAgentTitlePending(task.Metadata) {
|
|
pending := agentTitlePendingPredicate(r.db.DriverName())
|
|
metadataMerge := pendingTaskMetadataMergeExpression(r.db.DriverName())
|
|
updateQuery = fmt.Sprintf(`
|
|
UPDATE tasks SET workspace_id = ?, workflow_id = ?, workflow_step_id = ?,
|
|
title = CASE WHEN %s THEN ? ELSE title END,
|
|
description = ?, state = ?, priority = ?, position = ?, wip_admitted = ?,
|
|
queued_for_step_id = ?, queued_at = ?,
|
|
metadata = %s,
|
|
parent_id = ?, updated_at = ?, origin = ?, project_id = ?, labels = ?, identifier = ?
|
|
WHERE id = ?
|
|
`, pending, metadataMerge)
|
|
}
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(updateQuery), task.WorkspaceID, task.WorkflowID, task.WorkflowStepID, task.Title, task.Description, task.State, task.Priority, task.Position, dialect.BoolToInt(task.WIPAdmitted), task.QueuedForStepID, task.QueuedAt, string(metadata), task.ParentID, task.UpdatedAt, task.Origin, task.ProjectID, task.Labels, task.Identifier, task.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("%w: %s", ErrTaskNotFound, task.ID)
|
|
}
|
|
|
|
if err := r.recordStepTransition(ctx, tx, stepTransitionInput{
|
|
taskID: task.ID,
|
|
fromWorkflowID: fromWorkflowID,
|
|
fromWorkflowStepID: fromStepID,
|
|
toWorkflowID: task.WorkflowID,
|
|
toWorkflowStepID: task.WorkflowStepID,
|
|
occurredAt: task.UpdatedAt,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
return syncRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID)
|
|
}
|
|
|
|
// UpdateTaskWithWorkflowStepAdmission atomically moves a task into a workflow
|
|
// step. A limited full target stores the task in that destination as queued;
|
|
// it never rejects the move for WIP capacity.
|
|
func (r *Repository) UpdateTaskWithWorkflowStepAdmission(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
targetStepID string,
|
|
limit int,
|
|
) (bool, error) {
|
|
admitted, _, err := r.updateTaskWithWorkflowStepAdmission(ctx, task, targetStepID, limit, nil, false, "")
|
|
return admitted, err
|
|
}
|
|
|
|
// UpdateTaskWithWorkflowStepAdmissionAndState is the manual-move variant of
|
|
// UpdateTaskWithWorkflowStepAdmission. It keeps the destination admission,
|
|
// the state that applies after admission, and the queued source-exit marker
|
|
// in one transaction so a later full-row update cannot strand the move.
|
|
func (r *Repository) UpdateTaskWithWorkflowStepAdmissionAndState(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
targetStepID string,
|
|
limit int,
|
|
admittedState *v1.TaskState,
|
|
queueExitPending bool,
|
|
) (bool, error) {
|
|
admitted, _, err := r.updateTaskWithWorkflowStepAdmission(
|
|
ctx, task, targetStepID, limit, admittedState, queueExitPending, "",
|
|
)
|
|
return admitted, err
|
|
}
|
|
|
|
// UpdateTaskWithWorkflowStepAdmissionIfAtStep is the AC-46/48 compare-and-swap
|
|
// variant used by the workflow engine's guarded-transition re-evaluation
|
|
// apply path (see engine.TransitionStore.ApplyTransitionIfAtStep). The move
|
|
// is applied only if the task's persisted workflow_step_id still equals
|
|
// expectedStepID when read inside this transaction, after the workspace row
|
|
// lock — so on PostgreSQL a concurrent admission for the same workspace
|
|
// cannot race between the check and the write. applied=false means the
|
|
// precondition failed (the task already left expectedStepID by the time this
|
|
// ran) and is not an error; the task row is left untouched.
|
|
func (r *Repository) UpdateTaskWithWorkflowStepAdmissionIfAtStep(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
expectedStepID string,
|
|
targetStepID string,
|
|
limit int,
|
|
) (applied bool, err error) {
|
|
_, applied, err = r.updateTaskWithWorkflowStepAdmission(ctx, task, targetStepID, limit, nil, false, expectedStepID)
|
|
return applied, err
|
|
}
|
|
|
|
// rebaseTaskForStepAdmissionCAS applies the AC-46/48 compare-and-swap
|
|
// precondition for the CAS-guarded admission path: it reads the task's
|
|
// current step inside the transaction, after the workspace lock, via
|
|
// readTaskStepInTx so the read also takes its row-level FOR UPDATE lock on
|
|
// PostgreSQL. The workspace lock alone only serializes this method against
|
|
// other callers that take it (other admission calls); a plain writer like
|
|
// UpdateTask never takes the workspace lock and instead relies on that same
|
|
// FOR UPDATE lock via readTaskStepInTx, so a bare (lockless) SELECT here
|
|
// could read a stale expectedStepID, let UpdateTask commit a concurrent
|
|
// move, and then have this transaction's own write below overwrite it — a
|
|
// lost update. Taking the same row lock here blocks that writer until this
|
|
// transaction commits or rolls back, closing the race.
|
|
//
|
|
// This CAS path also reloads and rebases the full row (instead of writing
|
|
// the caller's snapshot) into task. The CAS caller (the workflow engine's
|
|
// guarded re-evaluation) can hold a task snapshot loaded well before this
|
|
// write — long enough for an unrelated concurrent edit (title, other
|
|
// metadata keys) to land in between. Writing that stale snapshot would
|
|
// silently clobber the concurrent edit. Only the fields this operation owns
|
|
// (workflow id, the deferred-moves metadata sub-map) are carried forward
|
|
// from the caller's request; everything else comes from the fresh row.
|
|
//
|
|
// The unconditional (expectedStepID == "") callers — manual/bulk/feeder
|
|
// moves — must NOT call this: their caller-built `task` already carries
|
|
// this operation's own field changes (Position, move-lifecycle metadata,
|
|
// …) that have no other source of truth, so rebasing from a fresh row would
|
|
// drop them instead of protecting them.
|
|
//
|
|
// Returns applied=false, err=nil when the CAS precondition fails (the task
|
|
// already left expectedStepID) — that is a lost race, not an error.
|
|
func (r *Repository) rebaseTaskForStepAdmissionCAS(
|
|
ctx context.Context,
|
|
tx *sql.Tx,
|
|
task *models.Task,
|
|
expectedStepID string,
|
|
now time.Time,
|
|
) (applied bool, err error) {
|
|
requestedWorkflowID := task.WorkflowID
|
|
requestedAppliedMoves := map[string]interface{}{}
|
|
if appliedMoves, ok := task.Metadata[models.MetaKeyAppliedDeferredMoves].(map[string]interface{}); ok {
|
|
for moveID, value := range appliedMoves {
|
|
requestedAppliedMoves[moveID] = value
|
|
}
|
|
}
|
|
|
|
_, currentStepID, found, err := r.readTaskStepInTx(ctx, tx, task.ID)
|
|
if err != nil {
|
|
return false, fmt.Errorf("read task step for admission CAS check: %w", err)
|
|
}
|
|
if !found {
|
|
return false, fmt.Errorf("%w: %s", ErrTaskNotFound, task.ID)
|
|
}
|
|
if currentStepID != expectedStepID {
|
|
return false, nil
|
|
}
|
|
currentTask, err := r.scanSingleTask(tx.QueryRowContext(ctx, r.db.Rebind(
|
|
`SELECT `+taskSelectColumns("t")+` FROM tasks t WHERE t.id = ?`), task.ID))
|
|
if err != nil {
|
|
return false, fmt.Errorf("reload task for admission CAS rebase: %w", err)
|
|
}
|
|
if requestedWorkflowID == "" {
|
|
requestedWorkflowID = currentTask.WorkflowID
|
|
}
|
|
*task = *currentTask
|
|
if requestedWorkflowID != "" {
|
|
task.WorkflowID = requestedWorkflowID
|
|
}
|
|
if task.Metadata == nil {
|
|
task.Metadata = map[string]interface{}{}
|
|
}
|
|
if len(requestedAppliedMoves) > 0 {
|
|
currentAppliedMoves := map[string]interface{}{}
|
|
if existing, ok := task.Metadata[models.MetaKeyAppliedDeferredMoves].(map[string]interface{}); ok {
|
|
for moveID, value := range existing {
|
|
currentAppliedMoves[moveID] = value
|
|
}
|
|
}
|
|
for moveID, value := range requestedAppliedMoves {
|
|
currentAppliedMoves[moveID] = value
|
|
}
|
|
task.Metadata[models.MetaKeyAppliedDeferredMoves] = currentAppliedMoves
|
|
}
|
|
task.UpdatedAt = now
|
|
return true, nil
|
|
}
|
|
|
|
func (r *Repository) updateTaskWithWorkflowStepAdmission(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
targetStepID string,
|
|
limit int,
|
|
admittedState *v1.TaskState,
|
|
queueExitPending bool,
|
|
expectedStepID string,
|
|
) (admitted bool, applied bool, err error) {
|
|
now := time.Now().UTC()
|
|
task.UpdatedAt = now
|
|
if task.Metadata == nil {
|
|
task.Metadata = map[string]interface{}{}
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
// Workspace row before workflow-step lock, matching createTask and the
|
|
// workspace cascade: the update path must not hold a step lock while the
|
|
// cascade holds the workspace and waits for that step (Postgres
|
|
// deadlock). The task's workspace is read from its row so the caller's
|
|
// model cannot bypass the ordering.
|
|
var workspaceID string
|
|
if err := tx.QueryRowContext(ctx, r.db.Rebind(`SELECT workspace_id FROM tasks WHERE id = ?`), task.ID).Scan(&workspaceID); err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// Preserve the ErrTaskNotFound sentinel callers relied on before
|
|
// the workspace read was introduced (a task deleted concurrently
|
|
// with a move is reachable).
|
|
return false, false, fmt.Errorf("%w: %s", ErrTaskNotFound, task.ID)
|
|
}
|
|
return false, false, fmt.Errorf("read task workspace for admission: %w", err)
|
|
}
|
|
if err := r.lockWorkspaceRowStdTx(ctx, tx, workspaceID); err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
// AC-46/48 compare-and-swap precondition, only for CAS callers (see
|
|
// rebaseTaskForStepAdmissionCAS for the full rationale). A mismatch means
|
|
// the task already left expectedStepID (lost the race) — that is
|
|
// reported as applied=false, not an error, and the transaction is rolled
|
|
// back untouched.
|
|
if expectedStepID != "" {
|
|
casApplied, err := r.rebaseTaskForStepAdmissionCAS(ctx, tx, task, expectedStepID, now)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
if !casApplied {
|
|
return false, false, nil
|
|
}
|
|
}
|
|
|
|
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, targetStepID); err != nil {
|
|
return false, false, err
|
|
}
|
|
occupants, err := r.countAdmittedInTx(ctx, tx, targetStepID, task.ID)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
admitted = task.IsEphemeral || limit <= 0 || occupants < limit
|
|
task.WorkflowStepID = targetStepID
|
|
if admitted {
|
|
task.WIPAdmitted = !task.IsEphemeral
|
|
task.QueuedForStepID = ""
|
|
task.QueuedAt = nil
|
|
if admittedState != nil {
|
|
task.State = *admittedState
|
|
}
|
|
} else {
|
|
task.WIPAdmitted = false
|
|
task.QueuedForStepID = targetStepID
|
|
task.QueuedAt = &now
|
|
}
|
|
if queueExitPending {
|
|
if admitted {
|
|
delete(task.Metadata, models.MetaKeyQueuedMoveExitPending)
|
|
} else {
|
|
if _, exists := task.Metadata[models.MetaKeyQueuedMoveExitPending]; !exists {
|
|
task.Metadata[models.MetaKeyQueuedMoveExitPending] = true
|
|
}
|
|
}
|
|
}
|
|
metadata, err := json.Marshal(task.Metadata)
|
|
if err != nil {
|
|
metadata = []byte("{}")
|
|
}
|
|
if err := r.updateTaskTx(ctx, tx, task, metadata); err != nil {
|
|
return false, false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return false, false, err
|
|
}
|
|
return admitted, true, nil
|
|
}
|
|
|
|
// RemoveTaskMetadataKey removes one metadata key without replacing concurrent
|
|
// task fields. It returns whether the key was present and removed.
|
|
func (r *Repository) RemoveTaskMetadataKey(ctx context.Context, taskID, key string) (bool, error) {
|
|
return r.removeTaskMetadataKeyWithExecutor(ctx, r.db, taskID, key)
|
|
}
|
|
|
|
func (r *Repository) removeTaskMetadataKeyWithExecutor(
|
|
ctx context.Context,
|
|
exec taskSessionExecutor,
|
|
taskID, key string,
|
|
) (bool, error) {
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `
|
|
UPDATE tasks SET metadata = (CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END #- ARRAY[?]::text[])::text, updated_at = ?
|
|
WHERE id = ? AND jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ?) IS NOT NULL
|
|
`
|
|
} else {
|
|
query = `
|
|
UPDATE tasks SET metadata = json_remove(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?), updated_at = ?
|
|
WHERE id = ? AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?) IS NOT NULL
|
|
`
|
|
}
|
|
path := jsonPath(key)
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
path = key
|
|
}
|
|
result, err := exec.ExecContext(ctx, r.db.Rebind(query), path, time.Now().UTC(), taskID, path)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
return rows > 0, err
|
|
}
|
|
|
|
// ClaimTaskTitleSession atomically assigns the first eligible session as the
|
|
// owner of a pending title handoff. Repeated calls by that same session are
|
|
// idempotent; a different session observes the existing owner and returns
|
|
// false.
|
|
func (r *Repository) ClaimTaskTitleSession(ctx context.Context, taskID, sessionID string) (bool, bool, error) {
|
|
if sessionID == "" {
|
|
return false, false, nil
|
|
}
|
|
|
|
var query string
|
|
var args []interface{}
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `
|
|
UPDATE tasks
|
|
SET metadata = jsonb_set(
|
|
CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END,
|
|
ARRAY['agent_title_owner_session_id']::text[], ?::jsonb, true
|
|
)::text,
|
|
updated_at = ?
|
|
WHERE id = ?
|
|
AND jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, 'agent_title_pending') = 'true'::jsonb
|
|
AND jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, 'agent_title_owner_session_id') IS NULL
|
|
`
|
|
payload, err := json.Marshal(sessionID)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
args = []interface{}{string(payload), time.Now().UTC(), taskID}
|
|
} else {
|
|
query = `
|
|
UPDATE tasks
|
|
SET metadata = json_set(
|
|
CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END,
|
|
'$.agent_title_owner_session_id', json(?)
|
|
), updated_at = ?
|
|
WHERE id = ?
|
|
AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, '$.agent_title_pending') = 'true'
|
|
AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, '$.agent_title_owner_session_id') IS NULL
|
|
`
|
|
payload, err := json.Marshal(sessionID)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
args = []interface{}{string(payload), time.Now().UTC(), taskID}
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), args...)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
if rows, rowsErr := result.RowsAffected(); rowsErr == nil && rows > 0 {
|
|
return true, true, nil
|
|
}
|
|
|
|
task, err := r.GetTask(ctx, taskID)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
return models.IsAgentTitleOwner(task.Metadata, sessionID), false, nil
|
|
}
|
|
|
|
// SetTaskTitleIfPending replaces a provisional title and removes its pending
|
|
// and owner markers in one conditional write. The compare-and-set prevents two
|
|
// agent sessions (or a late agent call racing a human rename) from both winning.
|
|
func (r *Repository) SetTaskTitleIfPending(ctx context.Context, taskID, sessionID, title string) (bool, error) {
|
|
if sessionID == "" {
|
|
return false, nil
|
|
}
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `
|
|
UPDATE tasks
|
|
SET title = ?,
|
|
metadata = (CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END #- ARRAY['agent_title_pending']::text[] #- ARRAY['agent_title_owner_session_id']::text[])::text,
|
|
updated_at = ?
|
|
WHERE id = ?
|
|
AND jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ?) = 'true'::jsonb
|
|
AND jsonb_extract_path_text(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, 'agent_title_owner_session_id') = ?
|
|
`
|
|
} else {
|
|
query = `
|
|
UPDATE tasks
|
|
SET title = ?,
|
|
metadata = json_remove(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, '$.agent_title_pending', '$.agent_title_owner_session_id'),
|
|
updated_at = ?
|
|
WHERE id = ?
|
|
AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?) = 'true'
|
|
AND json_extract(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, '$.agent_title_owner_session_id') = ?
|
|
`
|
|
}
|
|
path := jsonPath(models.MetaKeyAgentTitlePending)
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
path = models.MetaKeyAgentTitlePending
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), title, time.Now().UTC(), taskID, path, sessionID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
return rows > 0, err
|
|
}
|
|
|
|
// SetTaskMetadataKey updates one metadata key without replacing concurrent
|
|
// task fields. It is used to restore a deferred launch after a failed launch.
|
|
func (r *Repository) SetTaskMetadataKey(ctx context.Context, taskID, key string, value interface{}) error {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `UPDATE tasks SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ? WHERE id = ?`
|
|
} else {
|
|
query = `UPDATE tasks SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ? WHERE id = ?`
|
|
}
|
|
path := key
|
|
if !dialect.IsPostgres(r.db.DriverName()) {
|
|
path = jsonPath(key)
|
|
}
|
|
_, err = r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(), taskID)
|
|
return err
|
|
}
|
|
|
|
// SetTaskMetadataKeyIfNoActiveSession writes one metadata key only when the
|
|
// task has no session in STARTING or RUNNING. This prevents a delayed failure
|
|
// callback from re-adding a marker after a newer launch has already started.
|
|
func (r *Repository) SetTaskMetadataKeyIfNoActiveSession(
|
|
ctx context.Context, taskID, key string, value interface{},
|
|
) (bool, error) {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `UPDATE tasks
|
|
SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ?
|
|
WHERE id = ?
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM task_sessions
|
|
WHERE task_id = ? AND state IN (?, ?)
|
|
)`
|
|
} else {
|
|
query = `UPDATE tasks
|
|
SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ?
|
|
WHERE id = ?
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM task_sessions
|
|
WHERE task_id = ? AND state IN (?, ?)
|
|
)`
|
|
}
|
|
path := key
|
|
if !dialect.IsPostgres(r.db.DriverName()) {
|
|
path = jsonPath(key)
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(),
|
|
taskID, taskID, models.TaskSessionStateStarting, models.TaskSessionStateRunning)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
return rows > 0, err
|
|
}
|
|
|
|
// SetTaskMetadataKeyIfNotArchived writes one metadata key atomically with the
|
|
// archived_at guard: the write only lands when the task row still has
|
|
// archived_at IS NULL, and reports whether it landed. Startup reconciliation
|
|
// uses it for the interrupted_at marker so an archive that commits between the
|
|
// guard read and the metadata write can never leave a marker on an archived
|
|
// task (the check and the write are one statement).
|
|
func (r *Repository) SetTaskMetadataKeyIfNotArchived(ctx context.Context, taskID, key string, value interface{}) (bool, error) {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `UPDATE tasks SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ? WHERE id = ? AND archived_at IS NULL`
|
|
} else {
|
|
query = `UPDATE tasks SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ? WHERE id = ? AND archived_at IS NULL`
|
|
}
|
|
path := key
|
|
if !dialect.IsPostgres(r.db.DriverName()) {
|
|
path = jsonPath(key)
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(), taskID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
return rows > 0, err
|
|
}
|
|
|
|
// SetTaskMetadataKeyIfPresent rewrites one metadata key only while that key is
|
|
// still present, and reports whether the write landed.
|
|
//
|
|
// It is the compare-and-swap counterpart to RemoveTaskMetadataKey: an editor
|
|
// that read a key, decided to change it, and then writes it back must not
|
|
// RE-CREATE the key if a concurrent claim removed it in between. Editing a
|
|
// deferred launch prompt is the case this exists for — a full-row UpdateTask
|
|
// there would resurrect a launch intent that a just-started task had already
|
|
// consumed, and the gate would then fire a second session.
|
|
func (r *Repository) SetTaskMetadataKeyIfPresent(ctx context.Context, taskID, key string, value interface{}) (bool, error) {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var query string
|
|
if dialect.IsPostgres(r.db.DriverName()) {
|
|
query = `UPDATE tasks SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ?
|
|
WHERE id = ? AND jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ?) IS NOT NULL`
|
|
} else {
|
|
query = `UPDATE tasks SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ?
|
|
WHERE id = ? AND json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?) IS NOT NULL`
|
|
}
|
|
path := key
|
|
if !dialect.IsPostgres(r.db.DriverName()) {
|
|
path = jsonPath(key)
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(), taskID, path)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
return rows > 0, err
|
|
}
|
|
|
|
func jsonPath(key string) string { return "$." + key }
|
|
|
|
func agentTitlePendingPredicate(driver string) string {
|
|
if dialect.IsPostgres(driver) {
|
|
return "jsonb_extract_path(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, 'agent_title_pending') = 'true'::jsonb"
|
|
}
|
|
return "json_type(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, '$.agent_title_pending') = 'true'"
|
|
}
|
|
|
|
func pendingTaskMetadataMergeExpression(driver string) string {
|
|
if dialect.IsPostgres(driver) {
|
|
return "(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END || (?::jsonb - 'agent_title_pending' - 'agent_title_owner_session_id'))::text"
|
|
}
|
|
return "json_patch(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, json_remove(?, '$.agent_title_pending', '$.agent_title_owner_session_id'))"
|
|
}
|
|
|
|
// DetachTask clears only the hierarchy fields involved in detachment. Keeping
|
|
// this as a targeted update prevents concurrent task edits from being replaced
|
|
// by a stale full-row write.
|
|
func (r *Repository) DetachTask(ctx context.Context, taskID string) (bool, error) {
|
|
result, err := r.db.ExecContext(
|
|
ctx,
|
|
r.db.Rebind(detachTaskQuery(r.db.DriverName())),
|
|
time.Now().UTC(),
|
|
taskID,
|
|
)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
rows, err := result.RowsAffected()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if rows > 0 {
|
|
return true, nil
|
|
}
|
|
|
|
var existingID string
|
|
if err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`SELECT id FROM tasks WHERE id = ?`), taskID).Scan(&existingID); err != nil {
|
|
if err == sql.ErrNoRows {
|
|
return false, fmt.Errorf("%w: %s", ErrTaskNotFound, taskID)
|
|
}
|
|
return false, err
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
func detachTaskQuery(driver string) string {
|
|
if dialect.IsPostgres(driver) {
|
|
return `
|
|
UPDATE tasks
|
|
SET parent_id = '',
|
|
metadata = CASE
|
|
WHEN jsonb_extract_path_text(
|
|
CASE WHEN metadata IS NULL OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END,
|
|
'workspace', 'mode'
|
|
) = 'inherit_parent'
|
|
THEN jsonb_set(metadata::jsonb, '{workspace,mode}', '"shared_group"'::jsonb, true)::text
|
|
ELSE metadata
|
|
END,
|
|
updated_at = ?
|
|
WHERE id = ? AND parent_id != ''
|
|
`
|
|
}
|
|
return `
|
|
UPDATE tasks
|
|
SET parent_id = '',
|
|
metadata = CASE
|
|
WHEN json_valid(metadata) THEN CASE
|
|
WHEN json_extract(metadata, '$.workspace.mode') = 'inherit_parent'
|
|
THEN json_set(metadata, '$.workspace.mode', 'shared_group')
|
|
ELSE metadata
|
|
END
|
|
ELSE metadata
|
|
END,
|
|
updated_at = ?
|
|
WHERE id = ? AND parent_id != ''
|
|
`
|
|
}
|
|
|
|
// UpdateTaskIfWorkflowStepHasCapacity updates a task inside the same write
|
|
// transaction that checks a WIP-limited target step's current occupancy.
|
|
func (r *Repository) UpdateTaskIfWorkflowStepHasCapacity(ctx context.Context, task *models.Task, targetStepID, excludeTaskID string, limit int) error {
|
|
metadata, err := json.Marshal(task.Metadata)
|
|
if err != nil {
|
|
metadata = []byte("{}")
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, targetStepID); err != nil {
|
|
return err
|
|
}
|
|
|
|
var occupants int
|
|
if err := tx.QueryRowContext(ctx, r.db.Rebind(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE workflow_step_id = ?
|
|
AND wip_admitted = 1
|
|
AND id != ?
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), targetStepID, excludeTaskID).Scan(&occupants); err != nil {
|
|
return err
|
|
}
|
|
if occupants >= limit {
|
|
return wfmodels.NewWIPLimitError(targetStepID, limit, occupants)
|
|
}
|
|
|
|
fromWorkflowID, fromStepID, _, err := r.readTaskStepInTx(ctx, tx, task.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// See updateTaskTx's comment: stamped after the transactional lock, not
|
|
// before BeginTx, so occurred_at reflects true commit-serialization order
|
|
// under concurrent Postgres callers.
|
|
task.UpdatedAt = time.Now().UTC()
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET workspace_id = ?, workflow_id = ?, workflow_step_id = ?, title = ?, description = ?, state = ?, priority = ?, position = ?, wip_admitted = ?, queued_for_step_id = ?, queued_at = ?, metadata = ?, parent_id = ?, updated_at = ?, origin = ?, project_id = ?, labels = ?, identifier = ?
|
|
WHERE id = ?
|
|
`), task.WorkspaceID, task.WorkflowID, task.WorkflowStepID, task.Title, task.Description, task.State, task.Priority, task.Position, dialect.BoolToInt(task.WIPAdmitted), task.QueuedForStepID, task.QueuedAt, string(metadata), task.ParentID, task.UpdatedAt, task.Origin, task.ProjectID, task.Labels, task.Identifier, task.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("%w: %s", ErrTaskNotFound, task.ID)
|
|
}
|
|
if err := r.recordStepTransition(ctx, tx, stepTransitionInput{
|
|
taskID: task.ID,
|
|
fromWorkflowID: fromWorkflowID,
|
|
fromWorkflowStepID: fromStepID,
|
|
toWorkflowID: task.WorkflowID,
|
|
toWorkflowStepID: task.WorkflowStepID,
|
|
occurredAt: task.UpdatedAt,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if err := syncRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
// PromoteQueuedTaskIfWorkflowStepHasCapacity atomically claims a queued task
|
|
// for a destination step. The queue marker is part of the UPDATE predicate so
|
|
// concurrent reconcilers cannot promote the same row twice.
|
|
func (r *Repository) PromoteQueuedTaskIfWorkflowStepHasCapacity(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
fromStepID,
|
|
destinationStepID string,
|
|
limit int,
|
|
) (bool, error) {
|
|
metadata, err := json.Marshal(task.Metadata)
|
|
if err != nil {
|
|
metadata = []byte("{}")
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
if err := lockWorkflowStepForCapacity(ctx, tx, r.db.DriverName(), r.db.Rebind, destinationStepID); err != nil {
|
|
return false, err
|
|
}
|
|
if limit > 0 {
|
|
var occupants int
|
|
if err := tx.QueryRowContext(ctx, r.db.Rebind(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE workflow_step_id = ?
|
|
AND wip_admitted = 1
|
|
AND id != ?
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), destinationStepID, task.ID).Scan(&occupants); err != nil {
|
|
return false, err
|
|
}
|
|
if occupants >= limit {
|
|
return false, nil
|
|
}
|
|
}
|
|
queuePredicate := `
|
|
AND workflow_step_id = ?
|
|
AND (queued_for_step_id = ? OR queued_for_step_id = '' OR queued_for_step_id IS NULL)`
|
|
predicateArgs := []interface{}{fromStepID, destinationStepID}
|
|
if fromStepID == destinationStepID {
|
|
// A same-step promotion is a claim on a visible queue row. Once the
|
|
// first reconciler admits it, the row has no queue marker and must not
|
|
// satisfy a second promotion attempt. The empty-marker form remains
|
|
// reserved for legacy feeder rows below.
|
|
queuePredicate = `
|
|
AND workflow_step_id = ?
|
|
AND wip_admitted = 0
|
|
AND queued_for_step_id = ?`
|
|
}
|
|
|
|
fromWorkflowID, _, _, err := r.readTaskStepInTx(ctx, tx, task.ID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
// See updateTaskTx's comment: stamped after the transactional lock, not
|
|
// before BeginTx, so occurred_at reflects true commit-serialization order
|
|
// under concurrent Postgres callers.
|
|
task.UpdatedAt = time.Now().UTC()
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET workspace_id = ?, workflow_id = ?, workflow_step_id = ?, title = ?, description = ?, state = ?, priority = ?, position = ?, wip_admitted = ?, queued_for_step_id = ?, queued_at = ?, metadata = ?, parent_id = ?, updated_at = ?, origin = ?, project_id = ?, labels = ?, identifier = ?
|
|
WHERE id = ?
|
|
`+queuePredicate+`
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), append([]interface{}{
|
|
task.WorkspaceID, task.WorkflowID, task.WorkflowStepID, task.Title, task.Description, task.State, task.Priority, task.Position, dialect.BoolToInt(task.WIPAdmitted), task.QueuedForStepID, task.QueuedAt, string(metadata), task.ParentID, task.UpdatedAt, task.Origin, task.ProjectID, task.Labels, task.Identifier, task.ID,
|
|
}, predicateArgs...)...)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return false, nil
|
|
}
|
|
if err := r.recordStepTransition(ctx, tx, stepTransitionInput{
|
|
taskID: task.ID,
|
|
fromWorkflowID: fromWorkflowID,
|
|
fromWorkflowStepID: fromStepID,
|
|
toWorkflowID: task.WorkflowID,
|
|
toWorkflowStepID: task.WorkflowStepID,
|
|
occurredAt: task.UpdatedAt,
|
|
}); err != nil {
|
|
return false, err
|
|
}
|
|
if err := syncRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID); err != nil {
|
|
return false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
// DeleteTask deletes a task by ID
|
|
func (r *Repository) DeleteTask(ctx context.Context, id string) error {
|
|
tx, err := r.db.BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
// Serialize with session/worktree creation FIRST (task-row lock), then
|
|
// capture the authoritative session set: a concurrent CreateTaskSession
|
|
// holds the same task-row barrier, so every session committed before this
|
|
// lock is visible to the capture and anything after blocks until the task
|
|
// row is gone. Capturing before the lock could use a stale set (a session
|
|
// created mid-flight would never be purged). The session capture must also
|
|
// precede the task-row DELETE because task_sessions cascades on deletion.
|
|
if err := r.lockTaskRowInTx(ctx, tx, id); err != nil {
|
|
return err
|
|
}
|
|
sessions, err := r.taskQueueSessionsInTx(ctx, tx, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`DELETE FROM tasks WHERE id = ?`), id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
if err := r.purgeTaskQueueInTx(ctx, tx, id, sessions); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
r.notifyTaskQueuePurged(ctx, id)
|
|
return nil
|
|
}
|
|
|
|
// ListTasks returns all non-archived, non-ephemeral tasks for a workflow
|
|
func (r *Repository) ListTasks(ctx context.Context, workflowID string) ([]*models.Task, error) {
|
|
ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTasks")
|
|
defer span.End()
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.workflow_id = ? AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC
|
|
`), workflowID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// CountTasksByWorkflow returns the number of non-archived, non-ephemeral tasks in a workflow
|
|
func (r *Repository) CountTasksByWorkflow(ctx context.Context, workflowID string) (int, error) {
|
|
var count int
|
|
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`SELECT COUNT(*) FROM tasks WHERE workflow_id = ? AND archived_at IS NULL AND is_ephemeral = 0`+andNotAutomationOrigin), workflowID).Scan(&count)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// CountTasksByWorkflowStep returns the number of non-archived, non-ephemeral tasks in a workflow step
|
|
func (r *Repository) CountTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) {
|
|
var count int
|
|
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`SELECT COUNT(*) FROM tasks WHERE workflow_step_id = ? AND archived_at IS NULL AND is_ephemeral = 0`+andNotAutomationOrigin), stepID).Scan(&count)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// CountTasksByWorkflowStepExcludingTask returns active, admitted WIP occupants
|
|
// in a workflow step, excluding the task currently being moved.
|
|
func (r *Repository) CountTasksByWorkflowStepExcludingTask(ctx context.Context, stepID, excludeTaskID string) (int, error) {
|
|
var count int
|
|
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE workflow_step_id = ?
|
|
AND id != ?
|
|
AND wip_admitted = 1
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), stepID, excludeTaskID).Scan(&count)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// CountAdmittedTasksByWorkflowStep returns the active WIP occupants of a
|
|
// workflow step, excluding visible queued overflow cards.
|
|
func (r *Repository) CountAdmittedTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) {
|
|
var count int
|
|
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE workflow_step_id = ?
|
|
AND wip_admitted = 1
|
|
AND archived_at IS NULL
|
|
AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
`), stepID).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
// NextPullCandidate returns the next active, visible task from a feeder step.
|
|
func (r *Repository) NextPullCandidate(ctx context.Context, stepID, excludeTaskID string) (*models.Task, error) {
|
|
excludeTaskIDs := []string(nil)
|
|
if excludeTaskID != "" {
|
|
excludeTaskIDs = append(excludeTaskIDs, excludeTaskID)
|
|
}
|
|
return r.NextPullCandidateExcluding(ctx, stepID, excludeTaskIDs)
|
|
}
|
|
|
|
// NextPullCandidateExcluding returns the next active, visible task from a
|
|
// feeder step, skipping any candidate IDs the caller already tried.
|
|
func (r *Repository) NextPullCandidateExcluding(ctx context.Context, stepID string, excludeTaskIDs []string) (*models.Task, error) {
|
|
args := []any{stepID}
|
|
excludeClause := ""
|
|
if len(excludeTaskIDs) > 0 {
|
|
placeholders := make([]string, 0, len(excludeTaskIDs))
|
|
for _, id := range excludeTaskIDs {
|
|
if id == "" {
|
|
continue
|
|
}
|
|
placeholders = append(placeholders, "?")
|
|
args = append(args, id)
|
|
}
|
|
if len(placeholders) > 0 {
|
|
excludeClause = " AND t.id NOT IN (" + strings.Join(placeholders, ", ") + ")"
|
|
}
|
|
}
|
|
row := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.workflow_step_id = ?
|
|
AND t.archived_at IS NULL
|
|
AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
AND (t.queued_for_step_id = '' OR t.queued_for_step_id IS NULL)
|
|
`+excludeClause+`
|
|
ORDER BY
|
|
t.position ASC,
|
|
CASE LOWER(COALESCE(t.priority, ''))
|
|
WHEN 'critical' THEN 0
|
|
WHEN 'high' THEN 1
|
|
WHEN 'medium' THEN 2
|
|
WHEN 'low' THEN 3
|
|
WHEN 'none' THEN 4
|
|
ELSE 4
|
|
END ASC,
|
|
t.created_at ASC,
|
|
t.id ASC
|
|
LIMIT 1
|
|
`), args...)
|
|
task, err := r.scanSingleTask(row)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
return task, err
|
|
}
|
|
|
|
// NextQueuedTaskForStepExcluding returns the next task queued for destination
|
|
// from a feeder. Empty queue destinations preserve legacy feeder pull behavior.
|
|
func (r *Repository) NextQueuedTaskForStepExcluding(ctx context.Context, feederStepID, destinationStepID string, excludeTaskIDs []string) (*models.Task, error) {
|
|
args := []any{feederStepID, destinationStepID}
|
|
excludeClause := ""
|
|
if len(excludeTaskIDs) > 0 {
|
|
placeholders := make([]string, 0, len(excludeTaskIDs))
|
|
for _, id := range excludeTaskIDs {
|
|
if id == "" {
|
|
continue
|
|
}
|
|
placeholders = append(placeholders, "?")
|
|
args = append(args, id)
|
|
}
|
|
if len(placeholders) > 0 {
|
|
excludeClause = " AND t.id NOT IN (" + strings.Join(placeholders, ", ") + ")"
|
|
}
|
|
}
|
|
row := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.workflow_step_id = ?
|
|
AND t.archived_at IS NULL
|
|
AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
AND (t.queued_for_step_id = '' OR t.queued_for_step_id IS NULL OR t.queued_for_step_id = ?)
|
|
`+excludeClause+`
|
|
ORDER BY
|
|
t.position ASC,
|
|
CASE LOWER(COALESCE(t.priority, ''))
|
|
WHEN 'critical' THEN 0
|
|
WHEN 'high' THEN 1
|
|
WHEN 'medium' THEN 2
|
|
WHEN 'low' THEN 3
|
|
WHEN 'none' THEN 4
|
|
ELSE 4
|
|
END ASC,
|
|
COALESCE(t.queued_at, t.created_at) ASC,
|
|
t.created_at ASC,
|
|
t.id ASC
|
|
LIMIT 1
|
|
`), args...)
|
|
task, err := r.scanSingleTask(row)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
return task, err
|
|
}
|
|
|
|
// ListChildren returns non-archived, non-ephemeral children of parentID.
|
|
// Returns an empty list when parentID is empty (so root tasks resolve to
|
|
// "no children" cleanly).
|
|
func (r *Repository) ListChildren(ctx context.Context, parentID string) ([]*models.Task, error) {
|
|
if parentID == "" {
|
|
return []*models.Task{}, nil
|
|
}
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.parent_id = ? AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC, t.id ASC
|
|
`), parentID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListChildCompletionRows returns active direct children with the compact
|
|
// fields needed for on_children_completed readiness and operation idempotency.
|
|
func (r *Repository) ListChildCompletionRows(ctx context.Context, parentID string) ([]models.ChildCompletionRow, error) {
|
|
if parentID == "" {
|
|
return []models.ChildCompletionRow{}, nil
|
|
}
|
|
var rows []models.ChildCompletionRow
|
|
err := r.ro.SelectContext(ctx, &rows, r.ro.Rebind(`
|
|
SELECT id, state, title, workflow_step_id, updated_at
|
|
FROM tasks
|
|
WHERE parent_id = ? AND archived_at IS NULL AND is_ephemeral = 0`+andNotAutomationOrigin+`
|
|
ORDER BY created_at ASC, id ASC
|
|
`), parentID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return rows, nil
|
|
}
|
|
|
|
// ListChildrenIncludingArchived returns every child task of parentID
|
|
// regardless of archived state. Used by the office task-handoffs
|
|
// unarchive cascade (phase 6) to walk a previously-archived descendant
|
|
// subtree.
|
|
func (r *Repository) ListChildrenIncludingArchived(ctx context.Context, parentID string) ([]*models.Task, error) {
|
|
if parentID == "" {
|
|
return []*models.Task{}, nil
|
|
}
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.parent_id = ? AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC, t.id ASC
|
|
`), parentID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ReparentDirectChildren swaps the parent_id of every row matching
|
|
// oldParentID (archived or not) to newParentID. Used by no-cascade
|
|
// delete so the soon-to-be-orphaned direct children become roots
|
|
// instead of pointing at a row that's about to vanish.
|
|
func (r *Repository) ReparentDirectChildren(ctx context.Context, oldParentID, newParentID string) error {
|
|
if oldParentID == "" {
|
|
return nil
|
|
}
|
|
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET parent_id = ?, updated_at = ?
|
|
WHERE parent_id = ?
|
|
`), newParentID, time.Now().UTC(), oldParentID)
|
|
return err
|
|
}
|
|
|
|
// ListSiblings returns non-archived, non-ephemeral sibling tasks. A task
|
|
// is a sibling when parent_id matches AND the parent_id is non-empty AND
|
|
// the workspace matches. Root tasks (empty parent_id) deliberately return
|
|
// an empty list so unrelated workspace roots don't surface as siblings.
|
|
func (r *Repository) ListSiblings(ctx context.Context, taskID string) ([]*models.Task, error) {
|
|
self, err := r.GetTask(ctx, taskID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if self == nil || self.ParentID == "" {
|
|
return []*models.Task{}, nil
|
|
}
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.parent_id = ?
|
|
AND t.workspace_id = ?
|
|
AND t.id != ?
|
|
AND t.archived_at IS NULL
|
|
AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC, t.id ASC
|
|
`), self.ParentID, self.WorkspaceID, self.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListTasksByWorkflowStep returns all non-archived, non-ephemeral tasks in a workflow step
|
|
func (r *Repository) ListTasksByWorkflowStep(ctx context.Context, workflowStepID string) ([]*models.Task, error) {
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.workflow_step_id = ? AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+` ORDER BY t.created_at ASC
|
|
`), workflowStepID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListQueuedTasks returns non-archived, non-ephemeral tasks that are waiting
|
|
// for admission into a workflow step. It is used by startup reconciliation.
|
|
func (r *Repository) ListQueuedTasks(ctx context.Context) ([]*models.Task, error) {
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.queued_for_step_id IS NOT NULL AND t.queued_for_step_id != ''
|
|
AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.queued_at ASC, t.created_at ASC, t.id ASC
|
|
`))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListTasksWithMetadataKey returns active, non-ephemeral tasks carrying a
|
|
// named metadata key. It is used by startup lifecycle recovery, where queue
|
|
// destination columns alone cannot find already-admitted work whose entry or
|
|
// source-exit side effect still needs to run.
|
|
func (r *Repository) ListTasksWithMetadataKey(ctx context.Context, key string) ([]*models.Task, error) {
|
|
var predicate, path string
|
|
if dialect.IsPostgres(r.ro.DriverName()) {
|
|
predicate = "jsonb_extract_path(CASE WHEN t.metadata IS NULL OR t.metadata = 'null' OR t.metadata = '' THEN '{}'::jsonb ELSE t.metadata::jsonb END, ?) IS NOT NULL"
|
|
path = key
|
|
} else {
|
|
predicate = "json_type(CASE WHEN t.metadata IS NULL OR t.metadata = 'null' OR t.metadata = '' THEN '{}' ELSE t.metadata END, ?) IS NOT NULL"
|
|
path = jsonPath(key)
|
|
}
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE `+predicate+`
|
|
AND t.archived_at IS NULL
|
|
AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.updated_at ASC, t.created_at ASC, t.id ASC
|
|
`), path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListTasksByWorkspace returns paginated tasks for a workspace with total count
|
|
// If query is non-empty, filters by task title, description, repository name, or repository path
|
|
// If includeArchived is false, archived tasks are excluded. If onlyArchived
|
|
// is true, only archived tasks are returned and it takes precedence over
|
|
// includeArchived.
|
|
// If includeEphemeral is false, ephemeral tasks are excluded
|
|
// If onlyEphemeral is true, only ephemeral tasks are returned
|
|
func (r *Repository) ListTasksByWorkspace(ctx context.Context, workspaceID, workflowID, repositoryID, query string, page, pageSize int, sort string, includeArchived, includeEphemeral, onlyEphemeral, excludeConfig bool) ([]*models.Task, int, error) {
|
|
return r.ListTasksByWorkspaceWithArchiveMode(ctx, workspaceID, workflowID, repositoryID, query, page, pageSize, sort, includeArchived, includeEphemeral, onlyEphemeral, excludeConfig, false)
|
|
}
|
|
|
|
// ListTasksByWorkspaceWithArchiveMode is the additive workspace-list contract
|
|
// used by the sidebar archive view. onlyArchived takes precedence over
|
|
// includeArchived when both are true.
|
|
func (r *Repository) ListTasksByWorkspaceWithArchiveMode(ctx context.Context, workspaceID, workflowID, repositoryID, query string, page, pageSize int, sort string, includeArchived, includeEphemeral, onlyEphemeral, excludeConfig, onlyArchived bool) ([]*models.Task, int, error) {
|
|
ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTasksByWorkspace")
|
|
defer span.End()
|
|
// Calculate offset
|
|
offset := (page - 1) * pageSize
|
|
if offset < 0 {
|
|
offset = 0
|
|
}
|
|
sort = usermodels.NormalizeTasksListSort(sort)
|
|
|
|
// Build filter conditions
|
|
filter := ""
|
|
if onlyEphemeral {
|
|
// Only ephemeral tasks
|
|
filter += " AND is_ephemeral = 1"
|
|
} else if !includeEphemeral {
|
|
// Exclude ephemeral tasks
|
|
filter += " AND is_ephemeral = 0"
|
|
}
|
|
// If includeEphemeral is true and onlyEphemeral is false, include both
|
|
// Automation runs are excluded regardless of the ephemeral toggles: they
|
|
// are hidden by provenance, and "include quick chats" is not a request to
|
|
// see them.
|
|
filter += andNotAutomationOrigin
|
|
|
|
if onlyArchived {
|
|
filter += " AND archived_at IS NOT NULL"
|
|
} else if !includeArchived {
|
|
filter += " AND archived_at IS NULL"
|
|
}
|
|
|
|
if excludeConfig {
|
|
filter += " AND " + excludeConfigModePredicate(r.ro.DriverName(), "metadata")
|
|
}
|
|
|
|
var rows *sql.Rows
|
|
var total int
|
|
var err error
|
|
|
|
if query == "" {
|
|
rows, total, err = r.queryAllTasks(ctx, workspaceID, filter, workflowID, repositoryID, pageSize, offset, sort)
|
|
} else {
|
|
rows, total, err = r.searchTasks(ctx, workspaceID, query, filter, workflowID, repositoryID, pageSize, offset, sort)
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
tasks, err := r.scanTasks(rows)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
return tasks, total, nil
|
|
}
|
|
|
|
// queryAllTasks fetches all tasks (no search) for a workspace with pagination.
|
|
func (r *Repository) queryAllTasks(ctx context.Context, workspaceID, taskFilter, workflowID, repositoryID string, pageSize, offset int, sort string) (*sql.Rows, int, error) {
|
|
args := []interface{}{workspaceID}
|
|
if workflowID != "" {
|
|
taskFilter += " AND workflow_id = ?"
|
|
args = append(args, workflowID)
|
|
}
|
|
if repositoryID != "" {
|
|
taskFilter += " AND id IN (SELECT task_id FROM task_repositories WHERE repository_id = ?)"
|
|
args = append(args, repositoryID)
|
|
}
|
|
var total int
|
|
if err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`SELECT COUNT(*) FROM tasks WHERE workspace_id = ?`+taskFilter), args...).Scan(&total); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.workspace_id = ?`+rewriteFilterForAlias(taskFilter, "t")+`
|
|
ORDER BY `+taskListOrderBy(r.ro.DriverName(), "t", sort)+`
|
|
LIMIT ? OFFSET ?
|
|
`), append(append([]interface{}{}, args...), pageSize, offset)...)
|
|
return rows, total, err
|
|
}
|
|
|
|
func taskListOrderBy(driver, alias, sort string) string {
|
|
prefix := alias + "."
|
|
switch sort {
|
|
case usermodels.TasksListSortUpdatedAsc:
|
|
return prefix + "updated_at ASC, " + taskTitleOrder(driver, prefix, "ASC") + ", " + prefix + "id ASC"
|
|
case usermodels.TasksListSortCreatedDesc:
|
|
return prefix + "created_at DESC, " + taskTitleOrder(driver, prefix, "ASC") + ", " + prefix + "id ASC"
|
|
case usermodels.TasksListSortCreatedAsc:
|
|
return prefix + "created_at ASC, " + taskTitleOrder(driver, prefix, "ASC") + ", " + prefix + "id ASC"
|
|
case usermodels.TasksListSortTitleAsc:
|
|
return taskTitleOrder(driver, prefix, "ASC") + ", " + prefix + "updated_at DESC, " + prefix + "id ASC"
|
|
case usermodels.TasksListSortTitleDesc:
|
|
return taskTitleOrder(driver, prefix, "DESC") + ", " + prefix + "updated_at DESC, " + prefix + "id ASC"
|
|
case usermodels.TasksListSortUpdatedDesc:
|
|
fallthrough
|
|
default:
|
|
return prefix + "updated_at DESC, " + taskTitleOrder(driver, prefix, "ASC") + ", " + prefix + "id ASC"
|
|
}
|
|
}
|
|
|
|
func taskTitleOrder(driver, prefix, direction string) string {
|
|
if dialect.IsPostgres(driver) {
|
|
return "LOWER(" + prefix + "title) " + direction + ", " + prefix + "title " + direction
|
|
}
|
|
return prefix + "title COLLATE NOCASE " + direction
|
|
}
|
|
|
|
// rewriteFilterForAlias prefixes bare column references in `filter` with
|
|
// the given alias. Only used for the small, controlled fragments built
|
|
// in queryAllTasks and searchTasks.
|
|
func rewriteFilterForAlias(filter, alias string) string {
|
|
if alias == "" {
|
|
return filter
|
|
}
|
|
out := filter
|
|
for _, col := range []string{"is_ephemeral", "archived_at", "metadata", "workflow_id", "origin"} {
|
|
// Replace " <col>" only when not already prefixed by `alias.`.
|
|
out = simplePrefixCol(out, col, alias)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// simplePrefixCol replaces standalone occurrences of column names in
|
|
// SQL fragments with their aliased form. Naïve string substitution but
|
|
// adequate for the filters built locally above.
|
|
func simplePrefixCol(s, col, alias string) string {
|
|
if s == "" {
|
|
return s
|
|
}
|
|
prefix := alias + "."
|
|
out := ""
|
|
i := 0
|
|
for i < len(s) {
|
|
idx := indexFrom(s, col, i)
|
|
if idx == -1 {
|
|
out += s[i:]
|
|
break
|
|
}
|
|
// Boundary check: char before must not be alphanumeric or '.'.
|
|
if idx > 0 {
|
|
c := s[idx-1]
|
|
if c == '.' || isWordByte(c) {
|
|
out += s[i : idx+len(col)]
|
|
i = idx + len(col)
|
|
continue
|
|
}
|
|
}
|
|
// Boundary after: char after must not be alphanumeric.
|
|
end := idx + len(col)
|
|
if end < len(s) && isWordByte(s[end]) {
|
|
out += s[i : idx+len(col)]
|
|
i = idx + len(col)
|
|
continue
|
|
}
|
|
out += s[i:idx] + prefix + col
|
|
i = end
|
|
}
|
|
return out
|
|
}
|
|
|
|
func indexFrom(s, sub string, from int) int {
|
|
for i := from; i+len(sub) <= len(s); i++ {
|
|
if s[i:i+len(sub)] == sub {
|
|
return i
|
|
}
|
|
}
|
|
return -1
|
|
}
|
|
|
|
func isWordByte(b byte) bool {
|
|
return (b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') || b == '_'
|
|
}
|
|
|
|
// searchTasks fetches tasks matching a search query for a workspace with pagination.
|
|
func (r *Repository) searchTasks(ctx context.Context, workspaceID, query, filter, workflowID, repositoryID string, pageSize, offset int, sort string) (*sql.Rows, int, error) {
|
|
searchPattern := "%" + query + "%"
|
|
like := dialect.Like(r.ro.DriverName())
|
|
|
|
// Reuse the same archive, ephemeral, and config predicates as the
|
|
// non-search path so count and page results cannot diverge by mode.
|
|
tFilter := rewriteFilterForAlias(filter, "t")
|
|
|
|
// Collect extra filter args in query-argument order
|
|
var extraArgs []interface{}
|
|
if workflowID != "" {
|
|
tFilter += " AND t.workflow_id = ?"
|
|
extraArgs = append(extraArgs, workflowID)
|
|
}
|
|
if repositoryID != "" {
|
|
tFilter += " AND tr.repository_id = ?"
|
|
extraArgs = append(extraArgs, repositoryID)
|
|
}
|
|
|
|
countQuery := fmt.Sprintf(`
|
|
SELECT COUNT(DISTINCT t.id) FROM tasks t
|
|
LEFT JOIN task_repositories tr ON t.id = tr.task_id
|
|
LEFT JOIN repositories r ON tr.repository_id = r.id
|
|
WHERE t.workspace_id = ?%s
|
|
AND (
|
|
t.title %s ? OR
|
|
t.description %s ? OR
|
|
r.name %s ? OR
|
|
r.local_path %s ?
|
|
)
|
|
`, tFilter, like, like, like, like)
|
|
countArgs := append(append([]interface{}{workspaceID}, extraArgs...), searchPattern, searchPattern, searchPattern, searchPattern)
|
|
var total int
|
|
if err := r.ro.QueryRowContext(ctx, r.ro.Rebind(countQuery), countArgs...).Scan(&total); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
selectQuery := taskSearchSelectQuery(r.ro.DriverName(), tFilter, like, sort)
|
|
selectArgs := append(append([]interface{}{}, countArgs...), pageSize, offset)
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(selectQuery), selectArgs...)
|
|
return rows, total, err
|
|
}
|
|
|
|
func taskSearchSelectQuery(driver, tFilter, like, sort string) string {
|
|
return fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM (
|
|
SELECT DISTINCT %s
|
|
FROM tasks t
|
|
LEFT JOIN task_repositories tr ON t.id = tr.task_id
|
|
LEFT JOIN repositories r ON tr.repository_id = r.id
|
|
WHERE t.workspace_id = ?%s
|
|
AND (
|
|
t.title %s ? OR
|
|
t.description %s ? OR
|
|
r.name %s ? OR
|
|
r.local_path %s ?
|
|
)
|
|
) task_search
|
|
ORDER BY %s
|
|
LIMIT ? OFFSET ?
|
|
`, taskProjectedColumns("task_search"), taskSelectColumns("t"), tFilter, like, like, like, like, taskListOrderBy(driver, "task_search", sort))
|
|
}
|
|
|
|
// scanSingleTask scans a single row into a Task.
|
|
func (r *Repository) scanSingleTask(row *sql.Row) (*models.Task, error) {
|
|
task := &models.Task{}
|
|
var metadata string
|
|
var archivedAt sql.NullTime
|
|
var queuedAt sql.NullTime
|
|
var identifier sql.NullString
|
|
var externalID sql.NullString
|
|
var externalIDSettledAt sql.NullTime
|
|
err := row.Scan(
|
|
&task.ID, &task.WorkspaceID, &task.WorkflowID, &task.WorkflowStepID,
|
|
&task.Title, &task.Description, &task.State, &task.Priority, &task.Position,
|
|
&task.WIPAdmitted, &task.QueuedForStepID, &queuedAt,
|
|
&metadata, &task.IsEphemeral, &task.ParentID, &task.Autopilot, &archivedAt, &task.ArchivedByCascadeID,
|
|
&task.CreatedAt, &task.UpdatedAt,
|
|
&task.AssigneeAgentProfileID, &task.Origin, &task.ProjectID,
|
|
&task.Labels, &identifier, &externalID, &externalIDSettledAt, &task.IsFromOffice,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if archivedAt.Valid {
|
|
task.ArchivedAt = &archivedAt.Time
|
|
}
|
|
if queuedAt.Valid {
|
|
task.QueuedAt = &queuedAt.Time
|
|
}
|
|
if identifier.Valid {
|
|
task.Identifier = identifier.String
|
|
}
|
|
if externalID.Valid {
|
|
task.ExternalID = externalID.String
|
|
}
|
|
if externalIDSettledAt.Valid {
|
|
task.ExternalIDSettledAt = &externalIDSettledAt.Time
|
|
}
|
|
_ = json.Unmarshal([]byte(metadata), &task.Metadata)
|
|
return task, nil
|
|
}
|
|
|
|
// scanTasks is a helper to scan task rows
|
|
func (r *Repository) scanTasks(rows *sql.Rows) ([]*models.Task, error) {
|
|
var result []*models.Task
|
|
for rows.Next() {
|
|
task := &models.Task{}
|
|
var metadata string
|
|
var archivedAt sql.NullTime
|
|
var queuedAt sql.NullTime
|
|
var identifier sql.NullString
|
|
var externalID sql.NullString
|
|
var externalIDSettledAt sql.NullTime
|
|
err := rows.Scan(
|
|
&task.ID, &task.WorkspaceID, &task.WorkflowID, &task.WorkflowStepID,
|
|
&task.Title, &task.Description, &task.State, &task.Priority, &task.Position,
|
|
&task.WIPAdmitted, &task.QueuedForStepID, &queuedAt,
|
|
&metadata, &task.IsEphemeral, &task.ParentID, &task.Autopilot, &archivedAt, &task.ArchivedByCascadeID,
|
|
&task.CreatedAt, &task.UpdatedAt,
|
|
&task.AssigneeAgentProfileID, &task.Origin, &task.ProjectID,
|
|
&task.Labels, &identifier, &externalID, &externalIDSettledAt, &task.IsFromOffice,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if archivedAt.Valid {
|
|
task.ArchivedAt = &archivedAt.Time
|
|
}
|
|
if queuedAt.Valid {
|
|
task.QueuedAt = &queuedAt.Time
|
|
}
|
|
if identifier.Valid {
|
|
task.Identifier = identifier.String
|
|
}
|
|
if externalID.Valid {
|
|
task.ExternalID = externalID.String
|
|
}
|
|
if externalIDSettledAt.Valid {
|
|
task.ExternalIDSettledAt = &externalIDSettledAt.Time
|
|
}
|
|
_ = json.Unmarshal([]byte(metadata), &task.Metadata)
|
|
result = append(result, task)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
// GetTasksByIDs fetches multiple tasks in a single query. Missing IDs are
|
|
// silently omitted; result order is not guaranteed, so callers that need a
|
|
// specific order should reorder by ID themselves.
|
|
func (r *Repository) GetTasksByIDs(ctx context.Context, ids []string) ([]*models.Task, error) {
|
|
if len(ids) == 0 {
|
|
return nil, nil
|
|
}
|
|
placeholders := make([]string, len(ids))
|
|
args := make([]interface{}, len(ids))
|
|
for i, id := range ids {
|
|
placeholders[i] = "?"
|
|
args[i] = id
|
|
}
|
|
query := fmt.Sprintf(`SELECT %s FROM tasks t WHERE t.id IN (%s)`,
|
|
taskSelectColumns("t"), strings.Join(placeholders, ","))
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ArchiveTask sets the archived_at timestamp on a task
|
|
func (r *Repository) ArchiveTask(ctx context.Context, id string) error {
|
|
now := time.Now().UTC()
|
|
tx, err := r.db.BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`UPDATE tasks SET archived_at = ?, updated_at = ? WHERE id = ?`), now, now, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
sessions, err := r.taskQueueSessionsInTx(ctx, tx, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := r.purgeTaskQueueInTx(ctx, tx, id, sessions); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
r.notifyTaskQueuePurged(ctx, id)
|
|
return nil
|
|
}
|
|
|
|
// ArchiveTaskIfActive is the CAS variant used by office task-handoffs
|
|
// cascade archives. The update only fires when the task is currently
|
|
// active (archived_at IS NULL); this lets the cascade walk all
|
|
// descendants and archive only the ones not already archived by an
|
|
// earlier (manual or cascade) archive. Returns whether the row was
|
|
// actually updated.
|
|
//
|
|
// cascadeID is stamped on the archived row so UnarchiveTaskByCascade
|
|
// can scope its restoration to exactly the tasks this cascade owned.
|
|
// Pass empty cascadeID to opt out of cascade tracking (single-task
|
|
// manual archive); the column will simply not get set.
|
|
func (r *Repository) ArchiveTaskIfActive(ctx context.Context, id, cascadeID string) (bool, error) {
|
|
now := time.Now().UTC()
|
|
tx, err := r.db.BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET archived_at = ?, archived_by_cascade_id = ?, updated_at = ?
|
|
WHERE id = ? AND archived_at IS NULL
|
|
`), now, cascadeID, now, id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return false, tx.Commit()
|
|
}
|
|
sessions, err := r.taskQueueSessionsInTx(ctx, tx, id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if err := r.purgeTaskQueueInTx(ctx, tx, id, sessions); err != nil {
|
|
return false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return false, err
|
|
}
|
|
r.notifyTaskQueuePurged(ctx, id)
|
|
return true, nil
|
|
}
|
|
|
|
// taskQueueSessionsInTx returns the task's authoritative session set (its
|
|
// task_sessions rows). Callers must capture it BEFORE any statement that
|
|
// deletes the task row: task_sessions cascades on task deletion, so a
|
|
// post-delete discovery returns nothing and a concurrent admission to an
|
|
// empty session could survive the purge.
|
|
func (r *Repository) taskQueueSessionsInTx(ctx context.Context, tx *sqlx.Tx, taskID string) ([]string, error) {
|
|
var sessions []string
|
|
rows, err := tx.QueryxContext(ctx, r.db.Rebind(`SELECT id FROM task_sessions WHERE task_id = ?`), taskID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list task purge sessions: %w", err)
|
|
}
|
|
for rows.Next() {
|
|
var sessionID string
|
|
if err := rows.Scan(&sessionID); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("scan task purge session: %w", err)
|
|
}
|
|
sessions = append(sessions, sessionID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
_ = rows.Close()
|
|
return nil, fmt.Errorf("iterate task purge sessions: %w", err)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, fmt.Errorf("close task purge sessions: %w", err)
|
|
}
|
|
return sessions, nil
|
|
}
|
|
|
|
func (r *Repository) purgeTaskQueueInTx(ctx context.Context, tx *sqlx.Tx, taskID string, sessions []string) error {
|
|
_, err := messagequeue.PurgeTaskInTransaction(ctx, tx, r.db, taskID, sessions)
|
|
if internaldb.IsMissingTableError(err) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// UnarchiveTaskByCascade clears archived_at + archived_by_cascade_id only
|
|
// when the row was archived by the named cascade. Manually-archived tasks
|
|
// (empty cascade id) and tasks owned by a different cascade are left
|
|
// untouched — fixing the resurrection bug where unarchiving a parent
|
|
// would also un-archive descendants the user had archived manually
|
|
// before the cascade ran.
|
|
func (r *Repository) UnarchiveTaskByCascade(ctx context.Context, id, cascadeID string) (bool, error) {
|
|
if cascadeID == "" {
|
|
return false, fmt.Errorf("cascadeID is required")
|
|
}
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET archived_at = NULL, archived_by_cascade_id = '', updated_at = ?
|
|
WHERE id = ? AND archived_by_cascade_id = ?
|
|
`), time.Now().UTC(), id, cascadeID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
return rows > 0, nil
|
|
}
|
|
|
|
// UnarchiveTask clears archived_at for a manually/legacy-archived task
|
|
// (no cascade stamp). The CAS guard on archived_by_cascade_id keeps a
|
|
// delayed manual unarchive from erasing a newer cascade archive that
|
|
// landed between the caller's read and this update — cascade-stamped
|
|
// rows are only restored via UnarchiveTaskByCascade. Returns whether a
|
|
// row was actually updated.
|
|
func (r *Repository) UnarchiveTask(ctx context.Context, id string) (bool, error) {
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET archived_at = NULL, archived_by_cascade_id = '', updated_at = ?
|
|
WHERE id = ? AND archived_at IS NOT NULL
|
|
AND (archived_by_cascade_id = '' OR archived_by_cascade_id IS NULL)
|
|
`), time.Now().UTC(), id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
return rows > 0, nil
|
|
}
|
|
|
|
// ListTasksForAutoArchive returns tasks eligible for auto-archiving based on workflow step settings
|
|
func (r *Repository) ListTasksForAutoArchive(ctx context.Context) ([]*models.Task, error) {
|
|
drv := r.ro.DriverName()
|
|
query := fmt.Sprintf(`
|
|
SELECT %s
|
|
FROM tasks t
|
|
JOIN workflow_steps ws ON ws.id = t.workflow_step_id
|
|
WHERE ws.auto_archive_after_hours > 0
|
|
AND t.archived_at IS NULL
|
|
AND t.updated_at <= %s
|
|
`, taskSelectColumns("t"), dialect.NowMinusHours(drv, "ws.auto_archive_after_hours"))
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListArchivedTasksWithActiveSessions returns the IDs of archived tasks that
|
|
// still have at least one task_sessions row in an active DB state
|
|
// (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT). This is the candidate list
|
|
// for the periodic reconciliation sweep (see
|
|
// service.StartArchivedSessionReconciliationLoop): finalizeCancelledSessions
|
|
// bounds its session-cancellation retry to a handful of attempts, so
|
|
// sustained SQLite writer contention can exhaust it and leave an archived
|
|
// task's sessions stuck active with no session.state_changed event ever
|
|
// delivered. The sweep re-derives this list every pass, so a later attempt
|
|
// recovers the task once the contention clears.
|
|
func (r *Repository) ListArchivedTasksWithActiveSessions(ctx context.Context) ([]string, error) {
|
|
var ids []string
|
|
err := r.ro.SelectContext(ctx, &ids, `
|
|
SELECT DISTINCT t.id
|
|
FROM tasks t
|
|
JOIN task_sessions ts ON ts.task_id = t.id
|
|
WHERE t.archived_at IS NOT NULL
|
|
AND ts.state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT')
|
|
`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return ids, nil
|
|
}
|
|
|
|
// ListExpiredQuickChatTasks returns quick-chat tasks whose last task/session
|
|
// activity is older than cutoff. Active sessions are excluded so in-use chats
|
|
// are never deleted by the idle sweeper.
|
|
func (r *Repository) ListExpiredQuickChatTasks(ctx context.Context, cutoff time.Time) ([]*models.Task, error) {
|
|
drv := r.ro.DriverName()
|
|
sessionActivity := "COALESCE(MAX(ts.updated_at), t.updated_at)"
|
|
lastActivity := dialect.GreatestTimestamp(drv, "t.updated_at", sessionActivity)
|
|
query := fmt.Sprintf(`
|
|
WITH candidates AS (
|
|
SELECT t.id, %s AS last_activity
|
|
FROM tasks t
|
|
LEFT JOIN task_sessions ts ON ts.task_id = t.id
|
|
WHERE t.is_ephemeral = 1
|
|
AND COALESCE(t.workflow_id, '') = ''
|
|
AND COALESCE(t.origin, '') != ?
|
|
AND %s
|
|
AND t.archived_at IS NULL
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM task_sessions active
|
|
WHERE active.task_id = t.id
|
|
AND active.state IN (?, ?)
|
|
)
|
|
GROUP BY t.id, t.updated_at
|
|
HAVING %s < ?
|
|
)
|
|
SELECT %s
|
|
FROM tasks t
|
|
JOIN candidates c ON c.id = t.id
|
|
ORDER BY c.last_activity ASC
|
|
`,
|
|
lastActivity,
|
|
excludeConfigModePredicate(drv, "t.metadata"),
|
|
lastActivity,
|
|
taskSelectColumns("t"),
|
|
)
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query),
|
|
models.TaskOriginAutomationRun,
|
|
models.TaskSessionStateRunning,
|
|
models.TaskSessionStateIdle,
|
|
cutoff,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// DeleteExpiredQuickChatTask deletes id only when it still matches the expired
|
|
// quick-chat predicate at delete time.
|
|
func (r *Repository) DeleteExpiredQuickChatTask(ctx context.Context, id string, cutoff time.Time) (bool, error) {
|
|
drv := r.db.DriverName()
|
|
sessionActivity := "COALESCE(MAX(ts.updated_at), t.updated_at)"
|
|
lastActivity := dialect.GreatestTimestamp(drv, "t.updated_at", sessionActivity)
|
|
query := fmt.Sprintf(`
|
|
WITH candidate AS (
|
|
SELECT t.id, %s AS last_activity
|
|
FROM tasks t
|
|
LEFT JOIN task_sessions ts ON ts.task_id = t.id
|
|
WHERE t.id = ?
|
|
AND t.is_ephemeral = 1
|
|
AND COALESCE(t.workflow_id, '') = ''
|
|
AND COALESCE(t.origin, '') != ?
|
|
AND %s
|
|
AND t.archived_at IS NULL
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM task_sessions active
|
|
WHERE active.task_id = t.id
|
|
AND active.state IN (?, ?)
|
|
)
|
|
GROUP BY t.id, t.updated_at
|
|
HAVING %s < ?
|
|
)
|
|
DELETE FROM tasks
|
|
WHERE id = ?
|
|
AND EXISTS (SELECT 1 FROM candidate)
|
|
`,
|
|
lastActivity,
|
|
excludeConfigModePredicate(drv, "t.metadata"),
|
|
lastActivity,
|
|
)
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(query),
|
|
id,
|
|
models.TaskOriginAutomationRun,
|
|
models.TaskSessionStateRunning,
|
|
models.TaskSessionStateIdle,
|
|
cutoff,
|
|
id,
|
|
)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
return rows > 0, nil
|
|
}
|
|
|
|
// isSafeMetadataKey reports whether s is a safe JSON metadata key to splice
|
|
// into a json_extract path. The key is concatenated into the SQL text (it
|
|
// cannot be a bind parameter inside the '$.<key>' path literal), so it must be
|
|
// constrained to a fixed identifier alphabet to keep the query injection-safe.
|
|
// Callers pass compile-time constants today (WatcherSource.WatchMetadataKey),
|
|
// but validating here keeps the repository safe regardless of caller.
|
|
func isSafeMetadataKey(s string) bool {
|
|
if s == "" {
|
|
return false
|
|
}
|
|
for _, c := range s {
|
|
switch {
|
|
case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '_':
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// CountOpenWatcherCreatedTasks returns the number of open watcher-created tasks
|
|
// for a single watch, identified by the task-metadata key the integration
|
|
// writes (metadataKey, e.g. "sentry_issue_watch_id") and the watch id. "Open"
|
|
// means non-archived AND not in a terminal state (COMPLETED, FAILED,
|
|
// CANCELLED). Distinct integrations use distinct metadata keys, so counts are
|
|
// naturally scoped per integration without this repository knowing which
|
|
// integrations exist — the caller supplies the key.
|
|
//
|
|
// An empty watchID returns (0, nil) — no watch to count. A malformed
|
|
// metadataKey (not a bare [A-Za-z0-9_] identifier) returns an error rather
|
|
// than silently counting nothing, so a wiring bug surfaces in the logs (the
|
|
// throttle gate fails open on the error).
|
|
func (r *Repository) CountOpenWatcherCreatedTasks(ctx context.Context, metadataKey, watchID string) (int, error) {
|
|
if watchID == "" {
|
|
return 0, nil
|
|
}
|
|
if !isSafeMetadataKey(metadataKey) {
|
|
return 0, fmt.Errorf("invalid watcher metadata key %q", metadataKey)
|
|
}
|
|
query := r.ro.Rebind(fmt.Sprintf(`
|
|
SELECT COUNT(*) FROM tasks
|
|
WHERE archived_at IS NULL
|
|
AND state NOT IN (?, ?, ?)
|
|
AND %s = ?
|
|
`, dialect.JSONExtract(r.ro.DriverName(), "metadata", metadataKey)))
|
|
var n int
|
|
if err := r.ro.QueryRowxContext(ctx, query,
|
|
v1.TaskStateCompleted, v1.TaskStateFailed, v1.TaskStateCancelled, watchID,
|
|
).Scan(&n); err != nil {
|
|
return 0, err
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// UpdateTaskState updates the state of a task
|
|
func (r *Repository) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error {
|
|
result, err := r.db.ExecContext(ctx, r.db.Rebind(`UPDATE tasks SET state = ?, updated_at = ? WHERE id = ?`), state, time.Now().UTC(), id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// UpdateTaskStateIfSessionState atomically ties a task-state write to the
|
|
// current state of one owning session and to the task remaining unarchived.
|
|
// The task-state predicate makes the returned old state authoritative for
|
|
// task.state_changed events; a concurrent task or session write retries with
|
|
// a fresh snapshot or makes this update a no-op.
|
|
func (r *Repository) UpdateTaskStateIfSessionState(
|
|
ctx context.Context,
|
|
taskID, sessionID string,
|
|
expectedSessionState models.TaskSessionState,
|
|
state v1.TaskState,
|
|
) (v1.TaskState, bool, error) {
|
|
return r.updateTaskStateIfSessionState(
|
|
ctx, taskID, sessionID, expectedSessionState, state, false,
|
|
)
|
|
}
|
|
|
|
// UpdateTaskStateIfPrimarySessionState additionally requires the named
|
|
// session to remain primary.
|
|
func (r *Repository) UpdateTaskStateIfPrimarySessionState(
|
|
ctx context.Context,
|
|
taskID, sessionID string,
|
|
expectedSessionState models.TaskSessionState,
|
|
state v1.TaskState,
|
|
) (v1.TaskState, bool, error) {
|
|
return r.updateTaskStateIfSessionState(
|
|
ctx, taskID, sessionID, expectedSessionState, state, true,
|
|
)
|
|
}
|
|
|
|
func (r *Repository) updateTaskStateIfSessionState(
|
|
ctx context.Context,
|
|
taskID, sessionID string,
|
|
expectedSessionState models.TaskSessionState,
|
|
state v1.TaskState,
|
|
requirePrimary bool,
|
|
) (v1.TaskState, bool, error) {
|
|
for attempt := range updateTaskStateIfNotArchivedMaxAttempts {
|
|
if attempt > 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return "", false, ctx.Err()
|
|
case <-time.After(updateTaskStateIfNotArchivedBackoff):
|
|
}
|
|
}
|
|
oldState, updated, retry, err := r.tryUpdateTaskStateIfSessionState(
|
|
ctx, taskID, sessionID, expectedSessionState, state, requirePrimary,
|
|
)
|
|
if err != nil || !retry {
|
|
return oldState, updated, err
|
|
}
|
|
}
|
|
return "", false, fmt.Errorf("update task state if session state: exceeded %d attempts for task %s",
|
|
updateTaskStateIfNotArchivedMaxAttempts, taskID)
|
|
}
|
|
|
|
func (r *Repository) tryUpdateTaskStateIfSessionState(
|
|
ctx context.Context,
|
|
taskID, sessionID string,
|
|
expectedSessionState models.TaskSessionState,
|
|
state v1.TaskState,
|
|
requirePrimary bool,
|
|
) (oldState v1.TaskState, updated, retry bool, err error) {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return "", false, false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
var archivedAt sql.NullTime
|
|
var currentSessionState models.TaskSessionState
|
|
var currentSessionIsPrimary bool
|
|
err = tx.QueryRowContext(ctx, r.db.Rebind(`
|
|
SELECT tasks.state, tasks.archived_at, task_sessions.state, task_sessions.is_primary
|
|
FROM tasks
|
|
JOIN task_sessions ON task_sessions.task_id = tasks.id
|
|
WHERE tasks.id = ? AND task_sessions.id = ?
|
|
`), taskID, sessionID).Scan(&oldState, &archivedAt, ¤tSessionState, ¤tSessionIsPrimary)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, false, err
|
|
}
|
|
if archivedAt.Valid || currentSessionState != expectedSessionState ||
|
|
(requirePrimary && !currentSessionIsPrimary) {
|
|
return oldState, false, false, nil
|
|
}
|
|
requirePrimaryValue := 0
|
|
if requirePrimary {
|
|
requirePrimaryValue = 1
|
|
}
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks
|
|
SET state = ?, updated_at = ?
|
|
WHERE id = ?
|
|
AND state = ?
|
|
AND archived_at IS NULL
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM task_sessions
|
|
WHERE task_sessions.id = ?
|
|
AND task_sessions.task_id = tasks.id
|
|
AND task_sessions.state = ?
|
|
AND (? = 0 OR task_sessions.is_primary = 1)
|
|
)
|
|
`), state, time.Now().UTC(), taskID, oldState, sessionID, expectedSessionState, requirePrimaryValue)
|
|
if err != nil {
|
|
if isRetryableStateRaceError(err) {
|
|
return oldState, false, true, nil
|
|
}
|
|
return "", false, false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
if err != nil {
|
|
return "", false, false, err
|
|
}
|
|
if rows == 0 {
|
|
return oldState, false, true, nil
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return "", false, false, err
|
|
}
|
|
return oldState, true, false, nil
|
|
}
|
|
|
|
// RestoreTaskMessageRollbackIfSessionState atomically restores the two task
|
|
// fields changed by message_task's on-turn-start preparation, but only while
|
|
// the owning session remains in the state restored by the same rollback. A
|
|
// coordinator cancellation therefore prevents a late dispatch-failure
|
|
// rollback from moving the task out of REVIEW or rewinding its workflow step.
|
|
func (r *Repository) RestoreTaskMessageRollbackIfSessionState(
|
|
ctx context.Context,
|
|
task *models.Task,
|
|
sessionID string,
|
|
expectedSessionState models.TaskSessionState,
|
|
) (bool, error) {
|
|
if task == nil {
|
|
return false, errors.New("restore task message rollback: task is nil")
|
|
}
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
fromWorkflowID, fromStepID, _, err := r.readTaskStepInTx(ctx, tx, task.ID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
// See updateTaskTx's comment: stamped after the transactional lock, not
|
|
// before BeginTx, so occurred_at reflects true commit-serialization order
|
|
// under concurrent Postgres callers.
|
|
updatedAt := time.Now().UTC()
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks
|
|
SET state = ?, workflow_step_id = ?, updated_at = ?
|
|
WHERE id = ?
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM task_sessions
|
|
WHERE task_sessions.id = ?
|
|
AND task_sessions.task_id = tasks.id
|
|
AND task_sessions.state = ?
|
|
)
|
|
`), task.State, task.WorkflowStepID, updatedAt, task.ID, sessionID, expectedSessionState)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
rows, err := result.RowsAffected()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if rows == 0 {
|
|
return false, tx.Commit()
|
|
}
|
|
// workflow_id is not part of this UPDATE — a rollback restore only moves
|
|
// the step, never the workflow — so to_workflow_id equals from_workflow_id.
|
|
if err := r.recordStepTransition(ctx, tx, stepTransitionInput{
|
|
taskID: task.ID,
|
|
fromWorkflowID: fromWorkflowID,
|
|
fromWorkflowStepID: fromStepID,
|
|
toWorkflowID: fromWorkflowID,
|
|
toWorkflowStepID: task.WorkflowStepID,
|
|
occurredAt: updatedAt,
|
|
}); err != nil {
|
|
return false, err
|
|
}
|
|
if err := syncRunnerInTx(ctx, tx, r.db.Rebind, task.WorkflowStepID, task.ID, task.AssigneeAgentProfileID); err != nil {
|
|
return false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return false, err
|
|
}
|
|
task.UpdatedAt = updatedAt
|
|
return true, nil
|
|
}
|
|
|
|
// UpdateTaskStateIfCurrentIn transitions state inside a transaction, re-checking
|
|
// the current state on write so concurrent handlers cannot clobber a task that
|
|
// moved out of allowed between read and update. The write is also scoped to
|
|
// archived_at IS NULL: if ArchiveTask commits between the caller's earlier
|
|
// archived-state guard and this call, the archived_at check in the UPDATE's
|
|
// WHERE clause (not just the state check) makes the no-op atomic, so a late
|
|
// runtime write can never resurrect an archived task's state.
|
|
func (r *Repository) UpdateTaskStateIfCurrentIn(
|
|
ctx context.Context, id string, state v1.TaskState, allowed []v1.TaskState,
|
|
) (v1.TaskState, bool, error) {
|
|
if len(allowed) == 0 {
|
|
return "", false, nil
|
|
}
|
|
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
var currentState v1.TaskState
|
|
err = tx.QueryRowContext(ctx, r.db.Rebind(`SELECT state FROM tasks WHERE id = ?`), id).Scan(¤tState)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
return "", false, err
|
|
}
|
|
if !taskStateInSet(currentState, allowed) {
|
|
return currentState, false, nil
|
|
}
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET state = ?, updated_at = ?
|
|
WHERE id = ? AND state = ? AND archived_at IS NULL
|
|
`), state, time.Now().UTC(), id, currentState)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
return currentState, false, nil
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return "", false, err
|
|
}
|
|
return currentState, true, nil
|
|
}
|
|
|
|
// updateTaskStateIfNotArchivedMaxAttempts bounds the optimistic-retry loop in
|
|
// UpdateTaskStateIfNotArchived. A retry only fires when another writer
|
|
// changed tasks.state (not archived_at) in the gap between this method's
|
|
// read and its write; real contention on a single task's state column is
|
|
// rare enough that a handful of attempts is generous headroom, not a
|
|
// meaningful latency risk. updateTaskStateIfNotArchivedBackoff is a fixed
|
|
// delay between attempts, giving the concurrent writer time to finish
|
|
// (commit or roll back) before the next attempt re-reads — without it, a
|
|
// tight spin-retry can exhaust every attempt in well under a millisecond,
|
|
// faster than any real concurrent write completes.
|
|
const (
|
|
updateTaskStateIfNotArchivedMaxAttempts = 8
|
|
updateTaskStateIfNotArchivedBackoff = 15 * time.Millisecond
|
|
)
|
|
|
|
// UpdateTaskStateIfNotArchived atomically transitions state unless the task
|
|
// is archived — no prior-state constraint, unlike UpdateTaskStateIfCurrentIn.
|
|
// IN_PROGRESS writes (unlike the REVIEW CAS) are legitimately reachable from
|
|
// many prior states, so there is no "allowed" set to check; the only
|
|
// invariant that must hold atomically is archived_at IS NULL. Closes the
|
|
// same TOCTOU window as UpdateTaskStateIfCurrentIn: if ArchiveTask commits
|
|
// between a caller's earlier archived-state guard and this call, the
|
|
// archived_at check inside the transaction (not just the caller's read)
|
|
// makes the no-op atomic. Returns the pre-update state and whether a row
|
|
// was modified.
|
|
//
|
|
// The UPDATE's WHERE clause also pins state = currentState (the value read
|
|
// moments earlier in the same transaction), so the returned "pre-update
|
|
// state" is never stale: without that pin, a concurrent state write landing
|
|
// between the SELECT and the UPDATE would still match archived_at IS NULL
|
|
// alone, silently clobbering the newer state while this method kept
|
|
// reporting the old (now wrong) value as the pre-update state — which the
|
|
// service layer publishes as the task.state_changed event's old_state
|
|
// (CodeRabbit review on PR #1706). Losing the race on state (rows==0 while
|
|
// the row is still unarchived) means state moved, not that the row is gone
|
|
// or archived, so this is optimistic-concurrency-retried rather than
|
|
// treated as a no-op — an IN_PROGRESS write has no prior-state restriction
|
|
// to honor, so it always still applies once the retry re-reads a fresh
|
|
// state.
|
|
func (r *Repository) UpdateTaskStateIfNotArchived(
|
|
ctx context.Context, id string, state v1.TaskState,
|
|
) (v1.TaskState, bool, error) {
|
|
for attempt := range updateTaskStateIfNotArchivedMaxAttempts {
|
|
if attempt > 0 {
|
|
select {
|
|
case <-ctx.Done():
|
|
return "", false, ctx.Err()
|
|
case <-time.After(updateTaskStateIfNotArchivedBackoff):
|
|
}
|
|
}
|
|
currentState, updated, retry, err := r.tryUpdateTaskStateIfNotArchived(ctx, id, state)
|
|
if err != nil || !retry {
|
|
return currentState, updated, err
|
|
}
|
|
}
|
|
return "", false, fmt.Errorf("update task state if not archived: exceeded %d attempts for task %s",
|
|
updateTaskStateIfNotArchivedMaxAttempts, id)
|
|
}
|
|
|
|
// tryUpdateTaskStateIfNotArchived is one attempt of the optimistic-retry loop
|
|
// above. retry=true means the state changed concurrently while the row
|
|
// stayed unarchived — the caller should re-read and try again.
|
|
func (r *Repository) tryUpdateTaskStateIfNotArchived(
|
|
ctx context.Context, id string, state v1.TaskState,
|
|
) (currentState v1.TaskState, updated, retry bool, err error) {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return "", false, false, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
|
|
var archivedAt sql.NullTime
|
|
err = tx.QueryRowContext(ctx, r.db.Rebind(`SELECT state, archived_at FROM tasks WHERE id = ?`), id).
|
|
Scan(¤tState, &archivedAt)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, false, fmt.Errorf("%w: %s", ErrTaskNotFound, id)
|
|
}
|
|
return "", false, false, err
|
|
}
|
|
if archivedAt.Valid {
|
|
return currentState, false, false, nil
|
|
}
|
|
|
|
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
|
UPDATE tasks SET state = ?, updated_at = ?
|
|
WHERE id = ? AND state = ? AND archived_at IS NULL
|
|
`), state, time.Now().UTC(), id, currentState)
|
|
if err != nil {
|
|
if isRetryableStateRaceError(err) {
|
|
// SQLite (under WAL) can refuse to upgrade this transaction's
|
|
// already-established read snapshot to a writer once another
|
|
// connection committed a conflicting write in between — it
|
|
// returns SQLITE_BUSY ("database is locked") on the UPDATE
|
|
// itself rather than a clean 0-rows-affected. Same underlying
|
|
// condition as the rows==0 branch below: retry with a fresh
|
|
// transaction/snapshot instead of surfacing a transient error.
|
|
return currentState, false, true, nil
|
|
}
|
|
return "", false, false, err
|
|
}
|
|
rows, _ := result.RowsAffected()
|
|
if rows == 0 {
|
|
// archived_at IS NULL was true moments ago (checked above, inside
|
|
// this same transaction) — a concurrent writer changed state, not
|
|
// archived_at. Retry with a fresh read rather than reporting a
|
|
// stale old-state/no-op.
|
|
return currentState, false, true, nil
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return "", false, false, err
|
|
}
|
|
return currentState, true, false, nil
|
|
}
|
|
|
|
// isRetryableStateRaceError reports whether err is a database-level
|
|
// contention error rather than a genuine failure: SQLite's WAL snapshot
|
|
// conflict (a write following a read in the same transaction, after
|
|
// another connection committed a conflicting write — surfaces as
|
|
// SQLITE_BUSY/"database is locked" on the write itself, not a clean
|
|
// 0-rows-affected), or a Postgres serialization/deadlock/lock-timeout
|
|
// error. UpdateTaskStateIfNotArchived's retry loop treats this the same as
|
|
// rows==0: the underlying condition (another writer changed the row) is
|
|
// exactly the one it already retries for.
|
|
func isRetryableStateRaceError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
var pgErr *pgconn.PgError
|
|
if errors.As(err, &pgErr) {
|
|
switch pgErr.Code {
|
|
case "40001", "40P01", "55P03": // serialization_failure, deadlock_detected, lock_not_available
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
s := err.Error()
|
|
return strings.Contains(s, "database is locked") || strings.Contains(s, "database table is locked")
|
|
}
|
|
|
|
func taskStateInSet(state v1.TaskState, allowed []v1.TaskState) bool {
|
|
for _, candidate := range allowed {
|
|
if state == candidate {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// ListTasksByProject returns all non-archived, non-ephemeral tasks for a project.
|
|
func (r *Repository) ListTasksByProject(ctx context.Context, projectID string) ([]*models.Task, error) {
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE t.project_id = ? AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC
|
|
`), projectID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListTasksByAssignee returns all non-archived, non-ephemeral tasks assigned to an agent.
|
|
// The lookup goes through the runner participant projection so it picks
|
|
// up both per-task overrides and step-primary fallbacks (ADR 0005 Wave F).
|
|
func (r *Repository) ListTasksByAssignee(ctx context.Context, agentInstanceID string) ([]*models.Task, error) {
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
|
|
SELECT `+taskSelectColumns("t")+`
|
|
FROM tasks t
|
|
WHERE `+runnerProjection("t")+` = ?
|
|
AND t.archived_at IS NULL AND t.is_ephemeral = 0`+andNotAutomationOriginT+`
|
|
ORDER BY t.created_at ASC
|
|
`), agentInstanceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// ListTaskTree returns a flat list of non-archived tasks for a workspace, suitable for
|
|
// building a tree using each task's ParentID field.
|
|
func (r *Repository) ListTaskTree(ctx context.Context, workspaceID string, filters models.TaskTreeFilters) ([]*models.Task, error) {
|
|
query := `SELECT ` + taskSelectColumns("t") + ` FROM tasks t WHERE t.workspace_id = ? AND t.archived_at IS NULL AND t.is_ephemeral = 0` + andNotAutomationOriginT
|
|
args := []interface{}{workspaceID}
|
|
|
|
if filters.ProjectID != "" {
|
|
query += " AND t.project_id = ?"
|
|
args = append(args, filters.ProjectID)
|
|
}
|
|
if filters.AssigneeID != "" {
|
|
query += " AND " + runnerProjection("t") + " = ?"
|
|
args = append(args, filters.AssigneeID)
|
|
}
|
|
if filters.WorkflowID != "" {
|
|
query += " AND t.workflow_id = ?"
|
|
args = append(args, filters.WorkflowID)
|
|
}
|
|
if filters.Origin != "" {
|
|
query += " AND t.origin = ?"
|
|
args = append(args, filters.Origin)
|
|
}
|
|
query += " ORDER BY t.created_at ASC"
|
|
|
|
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
return r.scanTasks(rows)
|
|
}
|
|
|
|
// IncrementTaskSequence atomically increments the workspace task_sequence and returns the new value.
|
|
func (r *Repository) IncrementTaskSequence(ctx context.Context, workspaceID string) (int, error) {
|
|
var seq int
|
|
err := r.db.QueryRowContext(ctx, r.db.Rebind(`
|
|
UPDATE workspaces SET task_sequence = task_sequence + 1
|
|
WHERE id = ?
|
|
RETURNING task_sequence
|
|
`), workspaceID).Scan(&seq)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("increment task sequence for workspace %s: %w", workspaceID, err)
|
|
}
|
|
return seq, nil
|
|
}
|
|
|
|
// GetWorkspaceTaskPrefix returns the task prefix and office workflow ID for a workspace.
|
|
func (r *Repository) GetWorkspaceTaskPrefix(ctx context.Context, workspaceID string) (prefix, officeWorkflowID string, err error) {
|
|
err = r.ro.QueryRowContext(ctx, r.ro.Rebind(`
|
|
SELECT COALESCE(task_prefix, 'KAN'), COALESCE(office_workflow_id, '')
|
|
FROM workspaces WHERE id = ?
|
|
`), workspaceID).Scan(&prefix, &officeWorkflowID)
|
|
return
|
|
}
|