From 0591936e1692e78668404952dc66d20d02f1962a Mon Sep 17 00:00:00 2001 From: Alassane <79718216+yattdev@users.noreply.github.com> Date: Sun, 30 Aug 2026 10:34:40 +0000 Subject: [PATCH] fix: make task creation atomic when a reference fails to resolve (#3148) Co-authored-by: ayattara Co-authored-by: Carlos Florencio --- .../handlers/create_task_atomicity_test.go | 393 +++++++++++++++++ .../backend/internal/mcp/handlers/handlers.go | 6 + .../task/service/create_task_references.go | 132 ++++++ .../internal/task/service/service_tasks.go | 398 ++++++++++++------ 4 files changed, 807 insertions(+), 122 deletions(-) create mode 100644 apps/backend/internal/mcp/handlers/create_task_atomicity_test.go create mode 100644 apps/backend/internal/task/service/create_task_references.go diff --git a/apps/backend/internal/mcp/handlers/create_task_atomicity_test.go b/apps/backend/internal/mcp/handlers/create_task_atomicity_test.go new file mode 100644 index 0000000000..22fe5fd6d3 --- /dev/null +++ b/apps/backend/internal/mcp/handlers/create_task_atomicity_test.go @@ -0,0 +1,393 @@ +package handlers + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/stretchr/testify/require" + + orchmodels "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/task/service" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// unknownReferenceUUID is the id from the incident report: syntactically valid, +// but a task_repositories join-row id rather than a repositories.id, so it +// resolves to nothing. +const unknownReferenceUUID = "1f63b0c4-892d-44ea-a30c-f2a0300298bb" + +type atomicityFixture struct { + svc *service.Service + blockers *memBlockerRepo + repo *sqliterepo.Repository + handlers *Handlers + workspaceID string + workflowID string +} + +type noOpContributionDestinationPreparer struct{} + +func (*noOpContributionDestinationPreparer) PrepareContributionDestination( + context.Context, + *service.CreateTaskRequest, + *models.Workflow, + []*models.Repository, +) error { + return nil +} + +func newAtomicityFixture(t *testing.T) *atomicityFixture { + t.Helper() + svc, repo := newTestTaskService(t) + blockers := &memBlockerRepo{} + svc.SetBlockerRepository(blockers) + ctx := context.Background() + workspaces, err := svc.ListWorkspaces(ctx) + require.NoError(t, err) + workflows, err := svc.ListWorkflows(ctx, workspaces[0].ID, false) + require.NoError(t, err) + return &atomicityFixture{ + svc: svc, + blockers: blockers, + repo: repo, + handlers: NewHandlers(svc, nil, nil, nil, nil, repo, repo, nil, nil, nil, nil, nil, testLogger(t)), + workspaceID: workspaces[0].ID, + workflowID: workflows[0].ID, + } +} + +// newRepository inserts a repository row directly. Service-level +// CreateRepository additionally validates that local_path is a real git +// checkout, which is beside the point here: these tests only need an id that +// resolves. +func (f *atomicityFixture) newRepository(t *testing.T, name string) *models.Repository { + t.Helper() + repository := &models.Repository{WorkspaceID: f.workspaceID, Name: name} + require.NoError(t, f.repo.CreateRepository(context.Background(), repository)) + return repository +} + +// createdTaskID creates a task through the handler and returns its id. +func (f *atomicityFixture) createdTaskID(t *testing.T, title string) string { + t.Helper() + resp := f.create(t, map[string]interface{}{"title": title}) + require.NotEqual(t, ws.MessageTypeError, resp.Type, "setup create failed: %s", string(resp.Payload)) + var created map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &created)) + id, _ := created["id"].(string) + require.NotEmpty(t, id) + return id +} + +func (f *atomicityFixture) taskCount(t *testing.T) int { + t.Helper() + tasks, err := f.svc.ListTasks(context.Background(), f.workflowID) + require.NoError(t, err) + return len(tasks) +} + +func (f *atomicityFixture) create(t *testing.T, overrides map[string]interface{}) *ws.Message { + return f.createWithContext(t, context.Background(), overrides) +} + +func (f *atomicityFixture) createWithContext( + t *testing.T, + ctx context.Context, + overrides map[string]interface{}, +) *ws.Message { + t.Helper() + payload := map[string]interface{}{ + "workspace_id": f.workspaceID, + "workflow_id": f.workflowID, + "title": "Task", + "description": "Do the thing", + "agent_profile_id": "profile-1", + "start_agent": false, + } + for k, v := range overrides { + payload[k] = v + } + resp, err := f.handlers.handleCreateTask(ctx, makeWSMessage(t, ws.ActionMCPCreateTask, payload)) + require.NoError(t, err) + require.NotNil(t, resp) + return resp +} + +// memBlockerRepo is an in-memory BlockerRepository. The SQLite task repository +// does not implement the interface (blocker rows are Office-owned), and these +// tests only need edges to be creatable and readable back. +type memBlockerRepo struct { + edges []*orchmodels.TaskBlocker + // failCreateAfter makes the (failCreateAfter+1)-th CreateTaskBlocker fail, + // standing in for a blocker deleted between pre-insert validation and the + // post-insert write. Zero disables the injection. + failCreateAfter int + creates int + cancelOnFailure context.CancelFunc + cleanupCtxErr error + cleanupDeadline bool +} + +func (m *memBlockerRepo) CreateTaskBlocker(_ context.Context, blocker *orchmodels.TaskBlocker) error { + m.creates++ + if m.failCreateAfter > 0 && m.creates > m.failCreateAfter { + if m.cancelOnFailure != nil { + m.cancelOnFailure() + } + return errors.New("blocker vanished") + } + m.edges = append(m.edges, blocker) + return nil +} + +// DeleteTaskBlockersForTask satisfies the optional taskDependencyCleaner seam +// the service uses to tear down edges for a deleted task. +func (m *memBlockerRepo) DeleteTaskBlockersForTask(ctx context.Context, taskID string) error { + m.cleanupCtxErr = ctx.Err() + _, m.cleanupDeadline = ctx.Deadline() + kept := m.edges[:0] + for _, edge := range m.edges { + if edge.TaskID != taskID && edge.BlockerTaskID != taskID { + kept = append(kept, edge) + } + } + m.edges = kept + return nil +} + +func (m *memBlockerRepo) ListTaskBlockers(_ context.Context, taskID string) ([]*orchmodels.TaskBlocker, error) { + var out []*orchmodels.TaskBlocker + for _, edge := range m.edges { + if edge.TaskID == taskID { + out = append(out, edge) + } + } + return out, nil +} + +func (m *memBlockerRepo) DeleteTaskBlocker(_ context.Context, taskID, blockerTaskID string) error { + for i, edge := range m.edges { + if edge.TaskID == taskID && edge.BlockerTaskID == blockerTaskID { + m.edges = append(m.edges[:i], m.edges[i+1:]...) + return nil + } + } + return nil +} + +func (m *memBlockerRepo) ListTasksBlockedBy(_ context.Context, blockerTaskID string) ([]string, error) { + var out []string + for _, edge := range m.edges { + if edge.BlockerTaskID == blockerTaskID { + out = append(out, edge.TaskID) + } + } + return out, nil +} + +func (m *memBlockerRepo) ListBlockersForTasks(_ context.Context, taskIDs []string) (map[string][]string, error) { + out := make(map[string][]string) + for _, id := range taskIDs { + for _, edge := range m.edges { + if edge.TaskID == id { + out[id] = append(out[id], edge.BlockerTaskID) + } + } + } + return out, nil +} + +func (m *memBlockerRepo) ListDependentsForTasks(_ context.Context, blockerTaskIDs []string) (map[string][]string, error) { + out := make(map[string][]string) + for _, id := range blockerTaskIDs { + for _, edge := range m.edges { + if edge.BlockerTaskID == id { + out[id] = append(out[id], edge.TaskID) + } + } + } + return out, nil +} + +// TestHandleCreateTask_UnknownRepositoryIDLeavesNoOrphan is the regression for +// the reported defect. The handler inserted the task row, THEN resolved the +// repository, THEN returned an error — so a caller that correctly retried on +// failure ended up with two cards. +// +// Asserting only the error is insufficient: the whole defect is the row that +// survives it, so the row count is asserted directly on both sides of the call. +func TestHandleCreateTask_UnknownRepositoryIDLeavesNoOrphan(t *testing.T) { + f := newAtomicityFixture(t) + before := f.taskCount(t) + + resp := f.create(t, map[string]interface{}{ + "repositories": []map[string]interface{}{{"repository_id": unknownReferenceUUID}}, + }) + + failure := errorPayload(t, resp) + require.Equal(t, before, f.taskCount(t), "failed create must leave no orphan task row") + + // The error must be actionable on its own: a caller who cannot read the + // backend log still has to be able to fix their call from this message. + require.Equal(t, ws.ErrorCodeValidation, failure.Code, "an unresolvable caller-supplied id is a validation failure") + require.NotEqual(t, "Failed to create task", failure.Message, "the bare message names neither field nor value") + require.Contains(t, failure.Message, "repositories[0].repository_id", "error must name the failing field") + require.Contains(t, failure.Message, unknownReferenceUUID, "error must name the offending value") +} + +// TestHandleCreateTask_UnknownBlockerLeavesNoOrphan covers the other reference +// resolved after the insert. blocked_by went through AddBlocker in +// finalizeCreatedTask, so an unknown blocker stranded a task exactly as an +// unknown repository did. +func TestHandleCreateTask_UnknownBlockerLeavesNoOrphan(t *testing.T) { + f := newAtomicityFixture(t) + before := f.taskCount(t) + + resp := f.create(t, map[string]interface{}{ + "blocked_by": []string{unknownReferenceUUID}, + }) + + failure := errorPayload(t, resp) + require.Equal(t, before, f.taskCount(t), "failed create must leave no orphan task row") + require.Equal(t, ws.ErrorCodeValidation, failure.Code) + require.Contains(t, failure.Message, "blocked_by[0]", "error must name the failing field") + require.Contains(t, failure.Message, unknownReferenceUUID, "error must name the offending value") +} + +// TestHandleCreateTask_UnknownRepositoryIDIsNotPartiallyApplied guards the +// mixed case: a good repository listed before a bad one must not leave the +// task, or the first association, behind. +func TestHandleCreateTask_UnknownRepositoryIDIsNotPartiallyApplied(t *testing.T) { + f := newAtomicityFixture(t) + good := f.newRepository(t, "good-repo") + before := f.taskCount(t) + + resp := f.create(t, map[string]interface{}{ + "repositories": []map[string]interface{}{ + {"repository_id": good.ID}, + {"repository_id": unknownReferenceUUID}, + }, + }) + + failure := errorPayload(t, resp) + require.Equal(t, before, f.taskCount(t), "a later bad reference must not strand the task") + require.Equal(t, ws.ErrorCodeValidation, failure.Code) + require.Contains(t, failure.Message, "repositories[1].repository_id", "error must name the failing index") +} + +// TestHandleCreateTask_UnknownRepositoryInTemplatedWorkflowIsValidation guards +// the contribution-destination preparation path. That path runs before the +// main reference resolver for templated workflows, so it must use the same +// caller-facing reference error when its repository lookup fails. +func TestHandleCreateTask_UnknownRepositoryInTemplatedWorkflowIsValidation(t *testing.T) { + f := newAtomicityFixture(t) + workflow, err := f.repo.GetWorkflow(context.Background(), f.workflowID) + require.NoError(t, err) + templateID := "templated-workflow" + workflow.WorkflowTemplateID = &templateID + require.NoError(t, f.repo.UpdateWorkflow(context.Background(), workflow)) + f.svc.SetContributionDestinationPreparer(&noOpContributionDestinationPreparer{}) + + resp := f.create(t, map[string]interface{}{ + "repositories": []map[string]interface{}{{"repository_id": unknownReferenceUUID}}, + }) + + failure := errorPayload(t, resp) + require.Equal(t, ws.ErrorCodeValidation, failure.Code) + require.Contains(t, failure.Message, "repositories[0].repository_id") + require.Contains(t, failure.Message, unknownReferenceUUID) + require.Equal(t, 0, f.taskCount(t), "failed create must leave no orphan task row") +} + +// TestHandleCreateTask_BlockedByWithDeferredLaunchStillSucceeds is the +// no-behaviour-change guard for the exact request shape that produced the +// orphan on the live board: blocked_by plus a deferred launch. Moving +// reference resolution ahead of the insert must not disturb it. +func TestHandleCreateTask_BlockedByWithDeferredLaunchStillSucceeds(t *testing.T) { + f := newAtomicityFixture(t) + blockerResp := f.create(t, map[string]interface{}{"title": "Blocker"}) + require.NotEqual(t, ws.MessageTypeError, blockerResp.Type, "setup create failed: %s", string(blockerResp.Payload)) + var blocker map[string]interface{} + require.NoError(t, json.Unmarshal(blockerResp.Payload, &blocker)) + blockerID, _ := blocker["id"].(string) + require.NotEmpty(t, blockerID) + + repository := f.newRepository(t, "repo") + + before := f.taskCount(t) + resp := f.create(t, map[string]interface{}{ + "title": "Blocked child", + "blocked_by": []string{blockerID}, + "start_agent": true, + "repositories": []map[string]interface{}{{"repository_id": repository.ID}}, + }) + require.NotEqual(t, ws.MessageTypeError, resp.Type, "create failed: %s", string(resp.Payload)) + + var created map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &created)) + require.Equal(t, before+1, f.taskCount(t), "a successful create must still insert exactly one task") + + // The repository association must still be persisted after the insert. + taskID, _ := created["id"].(string) + require.NotEmpty(t, taskID) + task, err := f.svc.GetTask(context.Background(), taskID) + require.NoError(t, err) + require.Len(t, task.Repositories, 1, "resolved repositories must still be written") + require.Equal(t, repository.ID, task.Repositories[0].RepositoryID) + + blockers, err := f.svc.GetBlockers(context.Background(), taskID) + require.NoError(t, err) + require.Equal(t, []string{blockerID}, blockers, "blocker edge must still be created") +} + +// TestHandleCreateTask_RollbackAlsoRemovesDependencyEdges covers the residual +// TOCTOU path, where a blocker passes pre-insert validation and is gone by the +// time the edge is written. +// +// blocked_by is written one edge at a time, so failing on the second entry +// leaves the first already persisted. task_blockers predates the tasks foreign +// key and nothing cascades, so a rollback that deleted only the task row would +// swap an orphan task for an orphan edge pointing at a task that no longer +// exists. +func TestHandleCreateTask_RollbackAlsoRemovesDependencyEdges(t *testing.T) { + f := newAtomicityFixture(t) + first := f.createdTaskID(t, "Blocker one") + second := f.createdTaskID(t, "Blocker two") + + before := f.taskCount(t) + f.blockers.failCreateAfter = 1 + + resp := f.create(t, map[string]interface{}{ + "title": "Blocked child", + "blocked_by": []string{first, second}, + }) + + errorPayload(t, resp) + require.Equal(t, before, f.taskCount(t), "failed create must leave no orphan task row") + require.Empty(t, f.blockers.edges, "rollback must also remove the edge that was already written") +} + +func TestHandleCreateTask_RollbackSurvivesCallerCancellation(t *testing.T) { + f := newAtomicityFixture(t) + first := f.createdTaskID(t, "Blocker one") + second := f.createdTaskID(t, "Blocker two") + + ctx, cancel := context.WithCancel(context.Background()) + f.blockers.failCreateAfter = 1 + f.blockers.cancelOnFailure = cancel + before := f.taskCount(t) + + resp := f.createWithContext(t, ctx, map[string]interface{}{ + "title": "Blocked child", + "blocked_by": []string{first, second}, + }) + + errorPayload(t, resp) + require.Equal(t, before, f.taskCount(t), "rollback must delete the partial task after caller cancellation") + require.NoError(t, f.blockers.cleanupCtxErr, "rollback cleanup must not inherit caller cancellation") + require.True(t, f.blockers.cleanupDeadline, "rollback cleanup must remain bounded") +} diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 4adef7a702..0b1b1dee24 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -1018,6 +1018,12 @@ func classifyCreateTaskError(err error) string { case errors.Is(err, service.ErrSubtaskDepthExceeded), errors.Is(err, service.ErrInvalidTaskWorkflow), errors.Is(err, service.ErrExternalIDInvalid), + // A reference the caller supplied that does not resolve is a + // validation failure, not an internal one. Classifying it as + // INTERNAL_ERROR discarded err.Error() and left the caller with a + // bare "Failed to create task" naming neither the field nor the + // offending value — the cause reached the backend log and nowhere else. + errors.Is(err, service.ErrTaskReferenceNotFound), isMCPWorkflowNotFoundError(err): return ws.ErrorCodeValidation default: diff --git a/apps/backend/internal/task/service/create_task_references.go b/apps/backend/internal/task/service/create_task_references.go new file mode 100644 index 0000000000..95177ba974 --- /dev/null +++ b/apps/backend/internal/task/service/create_task_references.go @@ -0,0 +1,132 @@ +package service + +import ( + "context" + "errors" + "fmt" + "time" + + "go.uber.org/zap" + + taskrepo "github.com/kandev/kandev/internal/task/repository" + "github.com/kandev/kandev/internal/task/repository/repoerrors" +) + +// ErrTaskReferenceNotFound marks a create rejected because a caller-supplied +// reference (a repository, a blocker) does not resolve. +// +// It exists so transports can classify the failure as VALIDATION rather than +// INTERNAL_ERROR. A bad id in the request is the caller's to fix, and telling +// them only "Failed to create task" — as this path used to — leaves the actual +// cause visible nowhere but the backend log, which most callers cannot read. +var ErrTaskReferenceNotFound = errors.New("task reference not found") + +// unknownRepositoryReferenceError describes a repositories[i].repository_id +// that does not resolve, naming both the field and the offending value. +// +// The hint is not decoration: a task's repositories[] entry exposes BOTH an +// `id` (the task_repositories join row) and a `repository_id` (the repository +// itself), and passing the former is the mistake this error most often +// reports. +func unknownRepositoryReferenceError(index int, repositoryID string) error { + return fmt.Errorf( + "%w: unknown repositories[%d].repository_id %q; no repository with that id exists in this workspace. "+ + "If you copied it from a task's repositories[] entry, pass repository_id (the repository) rather than id (the task-repository link)", + ErrTaskReferenceNotFound, index, repositoryID, + ) +} + +// unknownBlockerReferenceError describes a blocked_by entry that does not +// resolve to a task in this workspace. +func unknownBlockerReferenceError(index int, blockerID string) error { + return fmt.Errorf( + "%w: unknown blocked_by[%d] %q; no task with that id exists in this workspace", + ErrTaskReferenceNotFound, index, blockerID, + ) +} + +// validateBlockerReferences checks every blocked_by id BEFORE the task row is +// inserted. +// +// AddBlocker remains the authoritative validator for dependency edges (see +// AddDependency's single-validator contract); this is not a second one. It +// resolves only what can be known without the new task existing — existence, +// authorization, and workspace agreement — so the request is rejected while +// there is still nothing to roll back. Cycle detection is deliberately absent: +// a task that does not exist yet has no outgoing edges and cannot close a +// cycle, and self-reference is impossible against an id not yet assigned. +func (s *Service) validateBlockerReferences(ctx context.Context, req *CreateTaskRequest) error { + if len(req.BlockedBy) == 0 { + return nil + } + if s.blockers == nil { + return ErrDependencyRepositoryUnavailable + } + for index, blockerID := range req.BlockedBy { + if blockerID == "" { + return fmt.Errorf("%w: blocked_by[%d] is empty", ErrTaskReferenceNotFound, index) + } + if err := s.authorizeTaskID(ctx, blockerID); err != nil { + // Denials surface as not-found so a caller cannot probe for the + // existence of another user's task, matching authorizeDependencyPair. + if errors.Is(err, taskrepo.ErrTaskNotFound) { + return unknownBlockerReferenceError(index, blockerID) + } + return err + } + blocker, err := s.tasks.GetTask(ctx, blockerID) + if err != nil { + if errors.Is(err, taskrepo.ErrTaskNotFound) { + return unknownBlockerReferenceError(index, blockerID) + } + return fmt.Errorf("resolve blocked_by[%d]: %w", index, err) + } + if blocker == nil { + return unknownBlockerReferenceError(index, blockerID) + } + if req.WorkspaceID != "" && blocker.WorkspaceID != "" && blocker.WorkspaceID != req.WorkspaceID { + return fmt.Errorf( + "%w: blocked_by[%d] %q belongs to a different workspace", + ErrTaskReferenceNotFound, index, blockerID, + ) + } + } + return nil +} + +// rollbackPartialTask deletes a task whose post-insert writes failed, and +// returns the original error to surface. +// +// Pre-insert validation closes the ordinary case; this closes the TOCTOU +// remainder, where a reference was valid when checked and gone by the time it +// was written. It mirrors the rollback the MCP create handler already performs +// for remote-contribution association and workspace-policy attach. A failed +// rollback is logged, never substituted for the original error: the caller +// needs to know why the create failed, not why the cleanup did. +// Cleanup is detached from caller cancellation but remains bounded, because a +// canceled request can be the reason finalization failed in the first place. +// +// Dependency edges are removed first. blocked_by is written one edge at a +// time, so a failure on the second entry leaves the first already persisted, +// and task_blockers predates the tasks foreign key — nothing cascades. Rolling +// back only the task row would trade an orphan task for an orphan edge +// pointing at a task that no longer exists. +func (s *Service) rollbackPartialTask(ctx context.Context, taskID string, cause error) error { + rollbackCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + s.deleteDependencyEdgesForTask(rollbackCtx, taskID) + if err := s.tasks.DeleteTask(rollbackCtx, taskID); err != nil { + s.logger.Error("rollback delete failed; task left in inconsistent state", + zap.String("task_id", taskID), zap.Error(err)) + } + return cause +} + +// classifyRepositoryResolutionError converts a repository lookup failure into +// the caller-facing reference error, leaving anything else untouched. +func classifyRepositoryResolutionError(index int, repositoryID string, err error) error { + if errors.Is(err, repoerrors.ErrRepositoryNotFound) { + return unknownRepositoryReferenceError(index, repositoryID) + } + return err +} diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index f48cdd4895..d2ef5cfe30 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -202,7 +202,7 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (Creat return result, err } - task, err := s.prepareTaskForCreation(ctx, req, externalID) + prepared, err := s.prepareTaskForCreation(ctx, req, externalID) if err != nil { if found, ok := s.recoverFoundTaskAfterInsertFailure(ctx, req.WorkspaceID, externalID); ok { return found, nil @@ -210,7 +210,7 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (Creat return CreateTaskResult{}, err } - if err := s.createTaskWithCapacity(ctx, task); err != nil { + if err := s.createTaskWithCapacity(ctx, prepared.task); err != nil { if found, ok := s.recoverFoundTaskAfterInsertFailure(ctx, req.WorkspaceID, externalID); ok { return found, nil } @@ -218,13 +218,25 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (Creat return CreateTaskResult{}, err } - return s.finalizeCreatedTask(ctx, task, req) + return s.finalizeCreatedTask(ctx, prepared, req) +} + +// preparedTask carries everything CreateTask resolved BEFORE inserting the +// task row: the row itself, and the fully-resolved task-repository rows that +// only need a TaskID stamped once the insert succeeds. +// +// Resolving the repositories here rather than in finalizeCreatedTask is the +// whole point: a reference that does not resolve must fail while there is +// still no task row to strand. +type preparedTask struct { + task *models.Task + repositories []*models.TaskRepository } // prepareTaskForCreation runs create-sequence steps 4-5's non-write half: // required validation, workflow/step resolution, and office identifier // assignment, producing the in-memory task CreateTask is about to insert. -func (s *Service) prepareTaskForCreation(ctx context.Context, req *CreateTaskRequest, externalID string) (*models.Task, error) { +func (s *Service) prepareTaskForCreation(ctx context.Context, req *CreateTaskRequest, externalID string) (*preparedTask, error) { // Subtasks created without an explicit project inherit the parent's, so // office cost events (which copy tasks.project_id verbatim) attribute to // the same project as the rest of the tree instead of leaking. Runs first @@ -283,7 +295,38 @@ func (s *Service) prepareTaskForCreation(ctx context.Context, req *CreateTaskReq return nil, err } } - return task, nil + + return s.resolveTaskCreationReferences(ctx, req, task) +} + +// resolveTaskCreationReferences resolves every remaining caller-supplied +// reference, and runs LAST in prepareTaskForCreation: the earlier steps +// rewrite req.Repositories in place (preflightRepositorySelections and +// prepareContributionDestination both do), so resolution has to see the final +// list. It still runs before the insert, which is the point — anything that +// can reject the request on bad input has now run with nothing yet written. +// +// One consequence is deliberate. A github_url or local_path input resolves +// through find-or-create, so a repository row may now be created before the +// task insert rather than after it, and a create that then fails admission +// (WIP limit, capacity) leaves that row behind. It is NOT rolled back, and the +// asymmetry is the reason: a stranded task row duplicates a card and carries +// deferred_launch, while a repository row is an idempotent workspace-level +// entity that the caller's retry resolves straight back onto. Deleting it +// would be the more dangerous choice — find-or-create means a concurrent task +// can bind to the same row in between, and rolling back would then delete a +// repository somebody else is using. +func (s *Service) resolveTaskCreationReferences( + ctx context.Context, req *CreateTaskRequest, task *models.Task, +) (*preparedTask, error) { + if err := s.validateBlockerReferences(ctx, req); err != nil { + return nil, err + } + repositories, err := s.resolveTaskRepositoryRows(ctx, req.WorkspaceID, req.Repositories) + if err != nil { + return nil, err + } + return &preparedTask{task: task, repositories: repositories}, nil } func (s *Service) prepareContributionDestination(ctx context.Context, req *CreateTaskRequest) error { @@ -318,10 +361,10 @@ func (s *Service) loadContributionDestinationRepositories( } repository, err := s.repoEntities.GetRepository(ctx, input.RepositoryID) if err != nil { - return nil, fmt.Errorf("load repository %s for contribution destination: %w", input.RepositoryID, err) + return nil, classifyRepositoryResolutionError(index, input.RepositoryID, err) } if repository == nil || repository.WorkspaceID != req.WorkspaceID { - return nil, repoerrors.ErrRepositoryNotFound + return nil, unknownRepositoryReferenceError(index, input.RepositoryID) } repositories[index] = repository } @@ -331,16 +374,19 @@ func (s *Service) loadContributionDestinationRepositories( // finalizeCreatedTask runs create-sequence step 6, the required synchronous // post-create work, after task has been inserted: blocker relationships, // task repositories, the created-event publish, and the feeder-pull refresh. -func (s *Service) finalizeCreatedTask(ctx context.Context, task *models.Task, req *CreateTaskRequest) (CreateTaskResult, error) { - // Create blocker relationships if specified. +func (s *Service) finalizeCreatedTask(ctx context.Context, prepared *preparedTask, req *CreateTaskRequest) (CreateTaskResult, error) { + task := prepared.task + // Create blocker relationships if specified. Every id here was validated + // pre-insert, so a failure now means the blocker was deleted underneath + // us — rare, but it must not leave the half-built task behind. for _, blockerID := range req.BlockedBy { if err := s.AddBlocker(ctx, task.ID, blockerID); err != nil { - return CreateTaskResult{}, fmt.Errorf("add blocker %s: %w", blockerID, err) + return CreateTaskResult{}, s.rollbackPartialTask(ctx, task.ID, fmt.Errorf("add blocker %s: %w", blockerID, err)) } } - if err := s.createTaskRepositories(ctx, task.ID, req.WorkspaceID, req.Repositories); err != nil { - return CreateTaskResult{}, err + if err := s.persistTaskRepositoryRows(ctx, task.ID, prepared.repositories); err != nil { + return CreateTaskResult{}, s.rollbackPartialTask(ctx, task.ID, err) } // Load repositories into task for response @@ -810,117 +856,21 @@ func (s *Service) assignIdentifier(ctx context.Context, task *models.Task) error // createTaskRepositories creates task-repository associations, resolving local paths to repository IDs. func (s *Service) createTaskRepositories(ctx context.Context, taskID, workspaceID string, repositories []TaskRepositoryInput) error { - var repoByPath map[string]*models.Repository - for _, repoInput := range repositories { - if repoInput.RepositoryID == "" && repoInput.LocalPath != "" { - repos, err := s.repoEntities.ListRepositories(ctx, workspaceID) - if err != nil { - s.logger.Error("failed to list repositories", zap.Error(err)) - return err - } - repoByPath = make(map[string]*models.Repository, len(repos)) - for _, repo := range repos { - if repo.LocalPath == "" { - continue - } - repoByPath[repo.LocalPath] = repo - } - break - } + rows, err := s.resolveTaskRepositoryRows(ctx, workspaceID, repositories) + if err != nil { + return err } + return s.persistTaskRepositoryRows(ctx, taskID, rows) +} - seen := make(map[string]bool, len(repositories)) - for i, repoInput := range repositories { - if repoInput.RemoteContribution != nil { - if err := repoInput.RemoteContribution.Validate(); err != nil { - return fmt.Errorf("invalid remote contribution: %w", err) - } - if repoInput.CheckoutBranch == "" { - repoInput.CheckoutBranch = repoInput.RemoteContribution.HeadBranch - } - if repoInput.BaseBranch == "" { - repoInput.BaseBranch = repoInput.RemoteContribution.BaseBranch - } - if repoInput.CheckoutBranch != repoInput.RemoteContribution.HeadBranch || - repoInput.BaseBranch != repoInput.RemoteContribution.BaseBranch { - return fmt.Errorf("remote contribution branches do not match the resolved binding") - } - } - if repoInput.ContributionDestination != nil { - if err := repoInput.ContributionDestination.Validate(); err != nil { - return fmt.Errorf("invalid contribution destination: %w", err) - } - } - repositoryID, baseBranch, _, err := s.resolveRepoInput(ctx, workspaceID, repoInput, repoByPath) - if err != nil { - return err - } - if repositoryID == "" { - return fmt.Errorf("repository_id is required") - } - policy, err := s.resolveTaskRepositoryPolicy(ctx, repositoryID, repoInput) - if err != nil { - return err - } - if policy != nil { - if repoInput.RemoteContribution != nil { - if policy.BaseBranch != repoInput.RemoteContribution.BaseBranch { - return fmt.Errorf("remote contribution base branch %q does not match branch policy base branch %q", repoInput.RemoteContribution.BaseBranch, policy.BaseBranch) - } - } else if !repoInput.PreserveBaseBranch { - baseBranch = policy.BaseBranch - } - } - // Multi-branch validation: the same repository may appear multiple - // times in a task on different branches. Identity is - // (repository_id, base_branch, checkout_branch) — base_branch matters - // because the worktree executor anchors the branch there while - // checkout_branch stays empty, and the local-executor flow puts the - // branch in checkout_branch with base_branch anchored to default_branch. - // Both shapes must dedup; matching DB key is UNIQUE(task_id, - // repository_id, base_branch, checkout_branch). - dedupKey := repositoryID + "\x00" + baseBranch + "\x00" + repoInput.CheckoutBranch - if seen[dedupKey] { - label := s.repoDisplayLabel(ctx, repoInput, repositoryID) - branchLabel := repoInput.CheckoutBranch - if branchLabel == "" { - branchLabel = baseBranch - } - if branchLabel == "" { - return fmt.Errorf("repository %q is listed more than once for this task", label) - } - return fmt.Errorf("repository %q on branch %q is listed more than once for this task", label, branchLabel) - } - seen[dedupKey] = true - metadata := make(map[string]interface{}) - if prNum := resolvePRNumber(repoInput); prNum > 0 { - metadata["pr_number"] = prNum - } - if repoInput.RemoteContribution != nil { - if err := models.PutRemoteContribution(metadata, repoInput.RemoteContribution); err != nil { - return fmt.Errorf("persist remote contribution: %w", err) - } - } - if repoInput.ContributionDestination != nil { - if err := models.PutContributionDestination(metadata, repoInput.ContributionDestination); err != nil { - return fmt.Errorf("persist contribution destination: %w", err) - } - } - taskRepo := &models.TaskRepository{ - TaskID: taskID, - RepositoryID: repositoryID, - BaseBranch: baseBranch, - CheckoutBranch: repoInput.CheckoutBranch, - Position: i, - Metadata: metadata, - } - if policy != nil { - taskRepo.BranchPolicyID = policy.ID - taskRepo.BranchPolicyName = policy.Name - taskRepo.BranchPolicyBaseBranch = policy.BaseBranch - taskRepo.BranchPolicyBranchTemplate = policy.BranchTemplate - taskRepo.BranchPolicyPullRequestTarget = policy.PullRequestTarget - } +// persistTaskRepositoryRows writes already-resolved task-repository rows, +// stamping them with the owning task. Split from resolution so CreateTask can +// resolve BEFORE inserting the task row: every reference failure that used to +// surface here — after the insert, leaving an orphan task — is now raised by +// resolveTaskRepositoryRows while there is still nothing to orphan. +func (s *Service) persistTaskRepositoryRows(ctx context.Context, taskID string, rows []*models.TaskRepository) error { + for _, taskRepo := range rows { + taskRepo.TaskID = taskID if err := s.taskRepos.CreateTaskRepository(ctx, taskRepo); err != nil { s.logger.Error("failed to create task repository", zap.Error(err)) return err @@ -929,6 +879,207 @@ func (s *Service) createTaskRepositories(ctx context.Context, taskID, workspaceI return nil } +// resolveTaskRepositoryRows resolves every repository input to a complete +// TaskRepository row WITHOUT persisting anything and without needing a task to +// exist yet: TaskID is left empty for persistTaskRepositoryRows to stamp. +// +// This half performs all reference resolution, policy lookup, and duplicate +// detection, so it is the half that can fail on caller-supplied data. Keeping +// it free of writes is what lets CreateTask run it pre-insert. +func (s *Service) resolveTaskRepositoryRows( + ctx context.Context, workspaceID string, repositories []TaskRepositoryInput, +) ([]*models.TaskRepository, error) { + repoByPath, err := s.repositoriesByLocalPath(ctx, workspaceID, repositories) + if err != nil { + return nil, err + } + seen := make(map[string]bool, len(repositories)) + rows := make([]*models.TaskRepository, 0, len(repositories)) + for i, repoInput := range repositories { + row, err := s.resolveTaskRepositoryRow(ctx, workspaceID, i, repoInput, repoByPath) + if err != nil { + return nil, err + } + if err := s.claimRepositoryBranchSlot(ctx, seen, row, repoInput); err != nil { + return nil, err + } + rows = append(rows, row) + } + return rows, nil +} + +// repositoriesByLocalPath indexes the workspace's repositories by local path, +// and only when some input actually needs the lookup. +func (s *Service) repositoriesByLocalPath( + ctx context.Context, workspaceID string, repositories []TaskRepositoryInput, +) (map[string]*models.Repository, error) { + needed := false + for _, repoInput := range repositories { + if repoInput.RepositoryID == "" && repoInput.LocalPath != "" { + needed = true + break + } + } + if !needed { + return nil, nil + } + repos, err := s.repoEntities.ListRepositories(ctx, workspaceID) + if err != nil { + s.logger.Error("failed to list repositories", zap.Error(err)) + return nil, err + } + repoByPath := make(map[string]*models.Repository, len(repos)) + for _, repo := range repos { + if repo.LocalPath == "" { + continue + } + repoByPath[repo.LocalPath] = repo + } + return repoByPath, nil +} + +// resolveTaskRepositoryRow resolves a single input into a complete, unsaved +// TaskRepository row. +func (s *Service) resolveTaskRepositoryRow( + ctx context.Context, workspaceID string, index int, + repoInput TaskRepositoryInput, repoByPath map[string]*models.Repository, +) (*models.TaskRepository, error) { + repoInput, err := normalizeContributionBindings(repoInput) + if err != nil { + return nil, err + } + repositoryID, baseBranch, _, err := s.resolveRepoInput(ctx, workspaceID, repoInput, repoByPath) + if err != nil { + if repoInput.RepositoryID != "" { + return nil, classifyRepositoryResolutionError(index, repoInput.RepositoryID, err) + } + return nil, err + } + if repositoryID == "" { + return nil, fmt.Errorf("repository_id is required") + } + policy, err := s.resolveTaskRepositoryPolicy(ctx, repositoryID, repoInput) + if err != nil { + return nil, err + } + baseBranch, err = applyBranchPolicyBaseBranch(baseBranch, repoInput, policy) + if err != nil { + return nil, err + } + metadata, err := buildTaskRepositoryMetadata(repoInput) + if err != nil { + return nil, err + } + row := &models.TaskRepository{ + RepositoryID: repositoryID, + BaseBranch: baseBranch, + CheckoutBranch: repoInput.CheckoutBranch, + Position: index, + Metadata: metadata, + } + if policy != nil { + row.BranchPolicyID = policy.ID + row.BranchPolicyName = policy.Name + row.BranchPolicyBaseBranch = policy.BaseBranch + row.BranchPolicyBranchTemplate = policy.BranchTemplate + row.BranchPolicyPullRequestTarget = policy.PullRequestTarget + } + return row, nil +} + +// normalizeContributionBindings validates the remote-contribution and +// contribution-destination bindings and fills in the branches they imply. +func normalizeContributionBindings(repoInput TaskRepositoryInput) (TaskRepositoryInput, error) { + if repoInput.RemoteContribution != nil { + if err := repoInput.RemoteContribution.Validate(); err != nil { + return repoInput, fmt.Errorf("invalid remote contribution: %w", err) + } + if repoInput.CheckoutBranch == "" { + repoInput.CheckoutBranch = repoInput.RemoteContribution.HeadBranch + } + if repoInput.BaseBranch == "" { + repoInput.BaseBranch = repoInput.RemoteContribution.BaseBranch + } + if repoInput.CheckoutBranch != repoInput.RemoteContribution.HeadBranch || + repoInput.BaseBranch != repoInput.RemoteContribution.BaseBranch { + return repoInput, fmt.Errorf("remote contribution branches do not match the resolved binding") + } + } + if repoInput.ContributionDestination != nil { + if err := repoInput.ContributionDestination.Validate(); err != nil { + return repoInput, fmt.Errorf("invalid contribution destination: %w", err) + } + } + return repoInput, nil +} + +// applyBranchPolicyBaseBranch reconciles the resolved base branch with the +// repository's branch policy. +func applyBranchPolicyBaseBranch( + baseBranch string, repoInput TaskRepositoryInput, policy *models.RepositoryBranchPolicy, +) (string, error) { + if policy == nil { + return baseBranch, nil + } + if repoInput.RemoteContribution != nil { + if policy.BaseBranch != repoInput.RemoteContribution.BaseBranch { + return "", fmt.Errorf("remote contribution base branch %q does not match branch policy base branch %q", + repoInput.RemoteContribution.BaseBranch, policy.BaseBranch) + } + return baseBranch, nil + } + if repoInput.PreserveBaseBranch { + return baseBranch, nil + } + return policy.BaseBranch, nil +} + +// buildTaskRepositoryMetadata assembles the row's metadata blob. +func buildTaskRepositoryMetadata(repoInput TaskRepositoryInput) (map[string]interface{}, error) { + metadata := make(map[string]interface{}) + if prNum := resolvePRNumber(repoInput); prNum > 0 { + metadata["pr_number"] = prNum + } + if repoInput.RemoteContribution != nil { + if err := models.PutRemoteContribution(metadata, repoInput.RemoteContribution); err != nil { + return nil, fmt.Errorf("persist remote contribution: %w", err) + } + } + if repoInput.ContributionDestination != nil { + if err := models.PutContributionDestination(metadata, repoInput.ContributionDestination); err != nil { + return nil, fmt.Errorf("persist contribution destination: %w", err) + } + } + return metadata, nil +} + +// claimRepositoryBranchSlot enforces multi-branch uniqueness. The same +// repository may appear multiple times in a task on different branches. +// Identity is (repository_id, base_branch, checkout_branch) — base_branch +// matters because the worktree executor anchors the branch there while +// checkout_branch stays empty, and the local-executor flow puts the branch in +// checkout_branch with base_branch anchored to default_branch. Both shapes +// must dedup; matching DB key is UNIQUE(task_id, repository_id, base_branch, +// checkout_branch). +func (s *Service) claimRepositoryBranchSlot( + ctx context.Context, seen map[string]bool, row *models.TaskRepository, repoInput TaskRepositoryInput, +) error { + dedupKey := row.RepositoryID + "\x00" + row.BaseBranch + "\x00" + row.CheckoutBranch + if !seen[dedupKey] { + seen[dedupKey] = true + return nil + } + label := s.repoDisplayLabel(ctx, repoInput, row.RepositoryID) + branchLabel := row.CheckoutBranch + if branchLabel == "" { + branchLabel = row.BaseBranch + } + if branchLabel == "" { + return fmt.Errorf("repository %q is listed more than once for this task", label) + } + return fmt.Errorf("repository %q on branch %q is listed more than once for this task", label, branchLabel) +} + // repoDisplayLabel returns a human-readable label for a repository to surface // in the duplicate-repository error. It prefers owner/name parsed from the // input's GitHub URL, then the resolved repo entity's owner/name (or bare @@ -1019,7 +1170,10 @@ func (s *Service) resolveRepoInputID(ctx context.Context, workspaceID, repositor return "", "", false, fmt.Errorf("looking up repository %q: %w", repositoryID, lookupErr) } if repo == nil || repo.WorkspaceID != workspaceID { - return "", "", false, fmt.Errorf("repository %q does not belong to workspace %q", repositoryID, workspaceID) + // Wrapped so transports classify this as a caller-supplied reference + // problem (VALIDATION) rather than an internal failure; the message + // itself is unchanged. + return "", "", false, fmt.Errorf("%w: repository %q does not belong to workspace %q", ErrTaskReferenceNotFound, repositoryID, workspaceID) } replacementID, replacementCreated, replacementErr := s.safeRepositoryIDForTaskWorktree(ctx, workspaceID, repo) if replacementErr != nil {