fix: make task creation atomic when a reference fails to resolve (#3148)

Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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")
+}

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:

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
+}

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 {