diff --git a/apps/backend/internal/backendapp/helpers.go b/apps/backend/internal/backendapp/helpers.go index a6279de7e3..e7722dcc2b 100644 --- a/apps/backend/internal/backendapp/helpers.go +++ b/apps/backend/internal/backendapp/helpers.go @@ -764,6 +764,7 @@ func registerRoutes(p routeParams) { handoffDocSvc := taskservice.NewDocumentService(p.taskRepo, p.log) handoffSvc := taskservice.NewHandoffService(p.taskRepo, p.taskRepo, handoffDocSvc, p.officeRepo, p.officeRepo, p.log) + p.taskSvc.SetWorkspacePolicyAttacher(handoffSvc) handoffSvc.SetCommentReader(&officeCommentReaderAdapter{reader: p.officeRepo}) // Phase 6 wirings — materializer hook + disk cleaner. The // SessionWorktreeReader and WorkspaceCleaner interfaces are both diff --git a/apps/backend/internal/backendapp/helpers_test.go b/apps/backend/internal/backendapp/helpers_test.go index 9e5bb94b76..ef3a96aa9c 100644 --- a/apps/backend/internal/backendapp/helpers_test.go +++ b/apps/backend/internal/backendapp/helpers_test.go @@ -1957,6 +1957,15 @@ type bootStateTestHarness struct { userSvc *userservice.Service } +// testWorkspacePolicyAttacher keeps boot-state tests focused on route and +// service composition while satisfying the service's required attachment +// boundary. Production wiring installs HandoffService here. +type testWorkspacePolicyAttacher struct{} + +func (testWorkspacePolicyAttacher) AttachWorkspacePolicy(context.Context, string, string, taskservice.WorkspacePolicy) error { + return nil +} + func newBootStateTestServices(t *testing.T) (*taskservice.Service, *workflowservice.Service) { harness := newBootStateTestHarness(t) return harness.taskSvc, harness.workflowSvc @@ -2027,6 +2036,7 @@ func newBootStateTestHarness(t *testing.T) bootStateTestHarness { taskSvc.SetWorkspaceBootstrapper(taskRepo) taskSvc.SetWorkflowStepGetter(&workflowStepGetterAdapter{svc: workflowSvc}) taskSvc.SetStartStepResolver(&startStepResolverAdapter{svc: workflowSvc}) + taskSvc.SetWorkspacePolicyAttacher(testWorkspacePolicyAttacher{}) workflowSvc.SetWorkflowProvider(&workflowProviderAdapter{svc: taskSvc}) return bootStateTestHarness{ db: sqlxDB, diff --git a/apps/backend/internal/integration/integration_test.go b/apps/backend/internal/integration/integration_test.go index cc0f2eedd6..04b6c78530 100644 --- a/apps/backend/internal/integration/integration_test.go +++ b/apps/backend/internal/integration/integration_test.go @@ -47,6 +47,15 @@ type TestServer struct { cancelFunc context.CancelFunc } +// testWorkspacePolicyAttacher keeps this integration harness focused on the +// task/MCP contract while satisfying the service's required attachment +// boundary. Production wiring installs HandoffService here. +type testWorkspacePolicyAttacher struct{} + +func (testWorkspacePolicyAttacher) AttachWorkspacePolicy(context.Context, string, string, taskservice.WorkspacePolicy) error { + return nil +} + // NewTestServer creates a new test server with all components initialized func NewTestServer(t *testing.T) *TestServer { t.Helper() @@ -100,6 +109,7 @@ func NewTestServer(t *testing.T) *TestServer { taskSvc.SetWorkflowStepCreator(workflowSvc) taskSvc.SetWorkflowStepGetter(workflowSvc) taskSvc.SetWorkspaceBootstrapper(taskRepo) + taskSvc.SetWorkspacePolicyAttacher(testWorkspacePolicyAttacher{}) // Create WebSocket gateway gateway := gateways.NewGateway(log) diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index c3100a2cc2..75f1aeb8c7 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -321,6 +321,16 @@ type Handlers struct { agentPermissionSvc AgentPermissionService } +func (h *Handlers) releaseWorkspacePolicyAfterCreateRollback(ctx context.Context, taskID string) { + if h.handoffSvc == nil || taskID == "" { + return + } + if err := h.handoffSvc.ReleaseWorkspacePolicy(context.WithoutCancel(ctx), taskID, "create_rollback"); err != nil { + h.logger.Warn("rollback workspace membership cleanup failed", + zap.String("task_id", taskID), zap.Error(err)) + } +} + // NewHandlers creates new MCP handlers. func NewHandlers( taskSvc *service.Service, @@ -889,6 +899,7 @@ func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.M DeferredLaunch: deferredLaunch, StartAgent: startAgent, ExternalID: req.ExternalID, + WorkspacePolicy: &workspacePolicy, }) if err != nil { h.logger.Error("failed to create task", zap.Error(err)) @@ -927,6 +938,7 @@ func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.M h.logger.Error("rollback delete failed after missing task repository", zap.String("task_id", task.ID), zap.Error(delErr)) } + h.releaseWorkspacePolicyAfterCreateRollback(ctx, task.ID) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to attach remote contribution: task repository is missing", nil) } if err := h.remoteContributionSvc.Associate(ctx, req.WorkspaceID, identity.UserID, task.ID, task.Repositories[index].RepositoryID, resolution); err != nil { @@ -936,22 +948,11 @@ func (h *Handlers) handleCreateTask(ctx context.Context, msg *ws.Message) (*ws.M h.logger.Error("rollback delete failed after contribution association error", zap.String("task_id", task.ID), zap.Error(delErr)) } + h.releaseWorkspacePolicyAfterCreateRollback(ctx, task.ID) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to attach remote contribution: "+err.Error(), nil) } } - if h.handoffSvc != nil && workspacePolicy.NeedsAttachment() { - if attachErr := h.handoffSvc.AttachWorkspacePolicy(ctx, task.ID, req.ParentID, workspacePolicy); attachErr != nil { - h.logger.Error("attach workspace policy; rolling back task creation", - zap.String("task_id", task.ID), zap.Error(attachErr)) - if delErr := h.taskSvc.DeleteTask(ctx, task.ID); delErr != nil { - h.logger.Error("rollback delete failed; task left in inconsistent state", - zap.String("task_id", task.ID), zap.Error(delErr)) - } - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to attach workspace policy: "+attachErr.Error(), nil) - } - } - // Settlement (create-sequence step 7): after policy attach, before // auto-start dispatch. settled, survivor, settleErr := h.taskSvc.SettleExternalID(ctx, task.ID, task.ExternalID) diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index c0779624af..551eba96c9 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -41,6 +41,12 @@ import ( // newTestTaskService creates a real task service with a temporary file-backed SQLite DB for integration tests. // Returns the service and the raw repo (for seeding data). +type testWorkspacePolicyAttacher struct{} + +func (testWorkspacePolicyAttacher) AttachWorkspacePolicy(context.Context, string, string, service.WorkspacePolicy) error { + return nil +} + func newTestTaskService(t *testing.T) (*service.Service, *sqliterepo.Repository) { svc, repo, _ := newTestTaskServiceWithEventBus(t) return svc, repo @@ -101,6 +107,7 @@ func newTestTaskServiceWithEventBus(t *testing.T) (*service.Service, *sqliterepo Environments: repo, Reviews: repo, }, eventBus, log, service.RepositoryDiscoveryConfig{}) + svc.SetWorkspacePolicyAttacher(testWorkspacePolicyAttacher{}) return svc, repo, eventBus } @@ -140,6 +147,7 @@ func newTestTaskServiceWithWorkflow(t *testing.T) (*service.Service, *sqliterepo Environments: repo, Reviews: repo, }, eventBus, log, service.RepositoryDiscoveryConfig{}) + svc.SetWorkspacePolicyAttacher(testWorkspacePolicyAttacher{}) workflowSvc := workflowservice.NewService(workflowRepo, log) t.Cleanup(func() { _ = workflowSvc.Close() }) return svc, repo, workflowcontroller.NewController(workflowSvc), workflowRepo diff --git a/apps/backend/internal/mcp/handlers/handoff_handlers.go b/apps/backend/internal/mcp/handlers/handoff_handlers.go index c569145963..a24375f95e 100644 --- a/apps/backend/internal/mcp/handlers/handoff_handlers.go +++ b/apps/backend/internal/mcp/handlers/handoff_handlers.go @@ -17,6 +17,13 @@ import ( // of registering and returning broken endpoints. func (h *Handlers) SetHandoffService(svc *service.HandoffService) { h.handoffSvc = svc + if h.taskSvc != nil { + if svc == nil { + h.taskSvc.SetWorkspacePolicyAttacher(nil) + } else { + h.taskSvc.SetWorkspacePolicyAttacher(svc) + } + } } // handleListRelatedTasks dispatches mcp.list_related_tasks. diff --git a/apps/backend/internal/office/models/workspace_groups.go b/apps/backend/internal/office/models/workspace_groups.go index 36b8a4694d..7f61737eba 100644 --- a/apps/backend/internal/office/models/workspace_groups.go +++ b/apps/backend/internal/office/models/workspace_groups.go @@ -11,6 +11,7 @@ type WorkspaceGroup struct { ID string `json:"id" db:"id"` WorkspaceID string `json:"workspace_id" db:"workspace_id"` OwnerTaskID string `json:"owner_task_id" db:"owner_task_id"` + OwnershipGeneration int64 `json:"ownership_generation" db:"ownership_generation"` MaterializedPath string `json:"materialized_path,omitempty" db:"materialized_path"` MaterializedEnvironmentID string `json:"materialized_environment_id,omitempty" db:"materialized_environment_id"` MaterializedKind string `json:"materialized_kind" db:"materialized_kind"` diff --git a/apps/backend/internal/office/repository/sqlite/base_migrations.go b/apps/backend/internal/office/repository/sqlite/base_migrations.go index 0c1b1e6141..724543c3e5 100644 --- a/apps/backend/internal/office/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/office/repository/sqlite/base_migrations.go @@ -42,6 +42,8 @@ func (r *Repository) runMigrations() { r.migrateRunOutcome() r.migrateParentWakeIndexes() r.migrateParentWakeReceiptColumns() + r.migrate.Apply("task_workspace_groups.ownership_generation", + `ALTER TABLE task_workspace_groups ADD COLUMN ownership_generation INTEGER NOT NULL DEFAULT 1`) } // migrateContinuationScope adds runs.continuation_scope for databases diff --git a/apps/backend/internal/office/repository/sqlite/workspace_groups.go b/apps/backend/internal/office/repository/sqlite/workspace_groups.go index 75ac1a7211..cd877918f0 100644 --- a/apps/backend/internal/office/repository/sqlite/workspace_groups.go +++ b/apps/backend/internal/office/repository/sqlite/workspace_groups.go @@ -25,6 +25,7 @@ func (r *Repository) createWorkspaceGroupTables() error { id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, owner_task_id TEXT NOT NULL, + ownership_generation INTEGER NOT NULL DEFAULT 1, materialized_path TEXT NOT NULL DEFAULT '', materialized_environment_id TEXT NOT NULL DEFAULT '', materialized_kind TEXT NOT NULL, @@ -80,6 +81,9 @@ func (r *Repository) CreateWorkspaceGroup(ctx context.Context, g *models.Workspa if g.CreatedAt.IsZero() { g.CreatedAt = now } + if g.OwnershipGeneration == 0 { + g.OwnershipGeneration = 1 + } g.UpdatedAt = now if g.CleanupPolicy == "" { g.CleanupPolicy = models.WorkspaceCleanupPolicyNeverDelete @@ -92,13 +96,13 @@ func (r *Repository) CreateWorkspaceGroup(ctx context.Context, g *models.Workspa } _, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO task_workspace_groups ( - id, workspace_id, owner_task_id, materialized_path, + id, workspace_id, owner_task_id, ownership_generation, materialized_path, materialized_environment_id, materialized_kind, owned_by_kandev, cleanup_policy, cleanup_status, cleaned_at, cleanup_error, restore_status, restore_error, restore_config_json, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `), g.ID, g.WorkspaceID, g.OwnerTaskID, g.MaterializedPath, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `), g.ID, g.WorkspaceID, g.OwnerTaskID, g.OwnershipGeneration, g.MaterializedPath, g.MaterializedEnvironmentID, g.MaterializedKind, g.OwnedByKandev, g.CleanupPolicy, g.CleanupStatus, g.CleanedAt, g.CleanupError, g.RestoreStatus, g.RestoreError, g.RestoreConfigJSON, @@ -110,7 +114,7 @@ func (r *Repository) CreateWorkspaceGroup(ctx context.Context, g *models.Workspa func (r *Repository) GetWorkspaceGroup(ctx context.Context, id string) (*models.WorkspaceGroup, error) { var g models.WorkspaceGroup err := r.ro.GetContext(ctx, &g, r.ro.Rebind(` - SELECT id, workspace_id, owner_task_id, materialized_path, + SELECT id, workspace_id, owner_task_id, ownership_generation, materialized_path, materialized_environment_id, materialized_kind, owned_by_kandev, cleanup_policy, cleanup_status, cleaned_at, cleanup_error, restore_status, restore_error, restore_config_json, @@ -131,7 +135,7 @@ func (r *Repository) GetWorkspaceGroup(ctx context.Context, id string) (*models. func (r *Repository) ListWorkspaceGroupsByWorkspace(ctx context.Context, workspaceID string) ([]*models.WorkspaceGroup, error) { var groups []*models.WorkspaceGroup err := r.ro.SelectContext(ctx, &groups, r.ro.Rebind(` - SELECT id, workspace_id, owner_task_id, materialized_path, + SELECT id, workspace_id, owner_task_id, ownership_generation, materialized_path, materialized_environment_id, materialized_kind, owned_by_kandev, cleanup_policy, cleanup_status, cleaned_at, cleanup_error, restore_status, restore_error, restore_config_json, @@ -218,6 +222,57 @@ func (r *Repository) UpdateWorkspaceGroupCleanupStatus(ctx context.Context, id, return err } +// ClaimWorkspaceGroupCleanup reserves destructive cleanup for one ownership +// generation. The active-member predicate is part of the same write so a +// caller cannot race a membership admission between its read and this claim. +func (r *Repository) ClaimWorkspaceGroupCleanup(ctx context.Context, id string, generation int64) (bool, error) { + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE task_workspace_groups + SET cleanup_status = ?, cleanup_error = '', cleaned_at = NULL, updated_at = ? + WHERE id = ? + AND ownership_generation = ? + AND owned_by_kandev = ? + AND cleanup_policy = ? + AND cleanup_status IN (?, ?) + AND NOT EXISTS ( + SELECT 1 + FROM task_workspace_group_members member + JOIN tasks task ON task.id = member.task_id + WHERE member.workspace_group_id = task_workspace_groups.id + AND member.released_at IS NULL + AND task.archived_at IS NULL + ) + `), models.WorkspaceCleanupStatusPending, time.Now().UTC(), id, generation, true, + models.WorkspaceCleanupPolicyDeleteWhenLastMemberArchivedOrDel, + models.WorkspaceCleanupStatusActive, models.WorkspaceCleanupStatusFailed) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + return rows == 1, err +} + +// CompleteWorkspaceGroupCleanup records a result only while the claimed +// ownership generation is still current. +func (r *Repository) CompleteWorkspaceGroupCleanup( + ctx context.Context, + id string, + generation int64, + status, errStr string, + cleanedAt *time.Time, +) (bool, error) { + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE task_workspace_groups + SET cleanup_status = ?, cleanup_error = ?, cleaned_at = ?, updated_at = ? + WHERE id = ? AND ownership_generation = ? AND cleanup_status = ? + `), status, errStr, cleanedAt, time.Now().UTC(), id, generation, models.WorkspaceCleanupStatusPending) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + return rows == 1, err +} + // UpdateWorkspaceGroupRestoreStatus updates the restore_status (+ optional // error) and bumps updated_at. func (r *Repository) UpdateWorkspaceGroupRestoreStatus(ctx context.Context, id, status, errStr string) error { @@ -238,12 +293,35 @@ func (r *Repository) AddWorkspaceGroupMember(ctx context.Context, groupID, taskI if role == "" { role = models.WorkspaceMemberRoleMember } - _, err := r.db.ExecContext(ctx, r.db.Rebind(` + result, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT OR IGNORE INTO task_workspace_group_members ( workspace_group_id, task_id, role, created_at - ) VALUES (?, ?, ?, ?) - `), groupID, taskID, role, time.Now().UTC()) - return err + ) + SELECT id, ?, ?, ? + FROM task_workspace_groups + WHERE id = ? AND cleanup_status IN ('active', 'cleanup_failed') + `), taskID, role, time.Now().UTC(), groupID) + if err != nil { + return err + } + if rows, err := result.RowsAffected(); err != nil { + return err + } else if rows == 0 { + var exists bool + if err := r.db.QueryRowContext(ctx, r.db.Rebind(` + SELECT EXISTS ( + SELECT 1 FROM task_workspace_group_members + WHERE workspace_group_id = ? AND task_id = ? + ) + `), groupID, taskID).Scan(&exists); err != nil { + return err + } + if exists { + return nil + } + return fmt.Errorf("workspace group %s is not accepting members", groupID) + } + return nil } // ReleaseWorkspaceGroupMember stamps released_at + release_reason and @@ -328,7 +406,7 @@ func (r *Repository) ListActiveWorkspaceGroupMembers(ctx context.Context, groupI func (r *Repository) GetWorkspaceGroupForTask(ctx context.Context, taskID string) (*models.WorkspaceGroup, error) { var g models.WorkspaceGroup err := r.ro.GetContext(ctx, &g, r.ro.Rebind(` - SELECT g.id, g.workspace_id, g.owner_task_id, g.materialized_path, + SELECT g.id, g.workspace_id, g.owner_task_id, g.ownership_generation, g.materialized_path, g.materialized_environment_id, g.materialized_kind, g.owned_by_kandev, g.cleanup_policy, g.cleanup_status, g.cleaned_at, g.cleanup_error, g.restore_status, g.restore_error, g.restore_config_json, diff --git a/apps/backend/internal/office/repository/sqlite/workspace_groups_test.go b/apps/backend/internal/office/repository/sqlite/workspace_groups_test.go index e7f1380291..01ce98d10e 100644 --- a/apps/backend/internal/office/repository/sqlite/workspace_groups_test.go +++ b/apps/backend/internal/office/repository/sqlite/workspace_groups_test.go @@ -108,6 +108,47 @@ func TestWorkspaceGroup_RawInsertDefaultsAreSafe(t *testing.T) { } } +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.4 +func TestWorkspaceGroupCleanupCompletionRejectsStaleGeneration(t *testing.T) { + repo := newWorkspaceGroupTestRepo(t) + ctx := context.Background() + group := &models.WorkspaceGroup{ + ID: "generation-fenced-cleanup", WorkspaceID: "ws-1", OwnerTaskID: "owner-task", + MaterializedKind: models.WorkspaceGroupKindPlainFolder, + OwnedByKandev: true, CleanupPolicy: models.WorkspaceCleanupPolicyDeleteWhenLastMemberArchivedOrDel, + } + if err := repo.CreateWorkspaceGroup(ctx, group); err != nil { + t.Fatalf("CreateWorkspaceGroup: %v", err) + } + claimed, err := repo.ClaimWorkspaceGroupCleanup(ctx, group.ID, 1) + if err != nil || !claimed { + t.Fatalf("ClaimWorkspaceGroupCleanup = %v, %v; want true", claimed, err) + } + if _, err := repo.ExecRaw(ctx, ` + UPDATE task_workspace_groups + SET ownership_generation = 2 + WHERE id = ? + `, group.ID); err != nil { + t.Fatalf("simulate stewardship transfer: %v", err) + } + completed, err := repo.CompleteWorkspaceGroupCleanup( + ctx, group.ID, 1, models.WorkspaceCleanupStatusCleaned, "", nil, + ) + if err != nil { + t.Fatalf("CompleteWorkspaceGroupCleanup: %v", err) + } + if completed { + t.Fatal("stale cleanup completion changed the replacement generation") + } + current, err := repo.GetWorkspaceGroup(ctx, group.ID) + if err != nil { + t.Fatalf("GetWorkspaceGroup: %v", err) + } + if current.OwnershipGeneration != 2 || current.CleanupStatus != models.WorkspaceCleanupStatusPending { + t.Fatalf("current group = generation %d status %q; want 2, cleanup_pending", current.OwnershipGeneration, current.CleanupStatus) + } +} + func TestWorkspaceGroup_ListByWorkspace(t *testing.T) { repo := newWorkspaceGroupTestRepo(t) ctx := context.Background() diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index d8a71f2a5e..1a93e872a9 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -77,9 +77,8 @@ type agentProfileRecentUseRecorder interface { } // SetHandoffService wires the office task-handoffs service used by the -// Kanban subtask path to attach workspace-group membership and the -// sequential blocker chain (handoffs phase 5). Optional — nil disables -// post-create attachment, matching the pre-handoffs behaviour. +// Kanban subtask path. The task service uses the same instance to attach +// workspace-group membership before any create route returns. // // Wiring a HandoffService also re-installs the per-user task guard on it. That // is not a convenience: this setter is what makes the archive / delete / unarchive @@ -91,8 +90,13 @@ type agentProfileRecentUseRecorder interface { // everywhere else. func (h *TaskHandlers) SetHandoffService(svc *service.HandoffService) { h.handoffSvc = svc - if svc != nil && h.service != nil { - svc.SetTaskAccessChecker(h.service.AuthorizeTaskAccess) + if h.service != nil { + if svc == nil { + h.service.SetWorkspacePolicyAttacher(nil) + } else { + svc.SetTaskAccessChecker(h.service.AuthorizeTaskAccess) + h.service.SetWorkspacePolicyAttacher(svc) + } } } diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index 0d0afa60a4..4310b61ebc 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -962,6 +962,7 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { ProjectID: body.ProjectID, Labels: labels, ExternalID: body.ExternalID, + WorkspacePolicy: &wsPolicy, }) if err != nil { handleNotFound(c, h.logger, err, "task not created") @@ -999,21 +1000,6 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { return } - if h.handoffSvc != nil && wsPolicy.NeedsAttachment() { - if attachErr := h.handoffSvc.AttachWorkspacePolicy(c.Request.Context(), task.ID, body.ParentID, wsPolicy); attachErr != nil { - h.logger.Error("attach workspace policy; rolling back task creation", - zap.String("task_id", task.ID), zap.Error(attachErr)) - if delErr := h.service.DeleteTask(c.Request.Context(), task.ID); delErr != nil { - h.logger.Error("rollback delete failed; task left in inconsistent state", - zap.String("task_id", task.ID), zap.Error(delErr)) - } - c.JSON(http.StatusInternalServerError, gin.H{ - "error": "failed to attach workspace policy: " + attachErr.Error(), - }) - return - } - } - if !h.commitFreshBranch(c, task.ID, task.Title, body.WorkspaceID, body.Repositories, repos) { return } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index d6df0cc0db..19b6a97557 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -2082,11 +2082,12 @@ const ( // It owns the workspace (worktree/container/sandbox) and the agentctl control server. // Multiple sessions can share the same TaskEnvironment. type TaskEnvironment struct { - ID string `json:"id"` - TaskID string `json:"task_id"` - ExecutorType string `json:"executor_type"` - ExecutorID string `json:"executor_id"` - ExecutorProfileID string `json:"executor_profile_id"` + ID string `json:"id"` + TaskID string `json:"task_id"` + OwnershipGeneration int64 `json:"ownership_generation"` + ExecutorType string `json:"executor_type"` + ExecutorID string `json:"executor_id"` + ExecutorProfileID string `json:"executor_profile_id"` // AgentExecutionID was removed: executors_running owns the execution<->session // mapping now. Read it via repo.GetExecutorRunningBySessionID(sessionID) when // needed (the orchestrator does this in service_turns.go for WorkspaceInfo). diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 4b2dc21af5..d2db0369f9 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -19,6 +19,7 @@ var ErrTaskParentMismatch = repoerrors.ErrTaskParentMismatch var ErrTaskPlanNotFound = repoerrors.ErrTaskPlanNotFound var ErrRepositoryNotFound = repoerrors.ErrRepositoryNotFound var ErrTaskEnvironmentNotFound = repoerrors.ErrTaskEnvironmentNotFound +var ErrTaskEnvironmentOwnershipChanged = repoerrors.ErrTaskEnvironmentOwnershipChanged var ErrWIPLimitExceeded = wfmodels.ErrWIPLimitExceeded var ErrExternalIDConflict = repoerrors.ErrExternalIDConflict diff --git a/apps/backend/internal/task/repository/repoerrors/errors.go b/apps/backend/internal/task/repository/repoerrors/errors.go index 0af56127cb..4f8aa9d77f 100644 --- a/apps/backend/internal/task/repository/repoerrors/errors.go +++ b/apps/backend/internal/task/repository/repoerrors/errors.go @@ -38,6 +38,10 @@ var ErrRepositoryBranchPoliciesExist = errors.New("repository branch policies al // ErrTaskEnvironmentNotFound reports that no task environment row matched the supplied id. var ErrTaskEnvironmentNotFound = errors.New("task environment not found") +// ErrTaskEnvironmentOwnershipChanged reports that an ownership transfer's +// expected owner or generation is no longer current. +var ErrTaskEnvironmentOwnershipChanged = errors.New("task environment ownership changed") + // ErrExternalIDConflict reports that a task insert lost the uniqueness race // on uniq_tasks_external_id — the TOCTOU backstop for the create sequence's // step-3 lookup (docs/specs/tasks/system-design/external-id-idempotency.md). Callers diff --git a/apps/backend/internal/task/repository/sqlite/base_migrations.go b/apps/backend/internal/task/repository/sqlite/base_migrations.go index 0849506365..8585e72e6f 100644 --- a/apps/backend/internal/task/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/task/repository/sqlite/base_migrations.go @@ -128,6 +128,7 @@ func (r *Repository) runMigrations() error { r.migrate.Apply("task_environments.materialization_session_id", `ALTER TABLE task_environments ADD COLUMN materialization_session_id TEXT DEFAULT ''`) r.migrate.Apply("task_environments.container_bootstrap_nonce_secret_id", `ALTER TABLE task_environments ADD COLUMN container_bootstrap_nonce_secret_id TEXT DEFAULT ''`) r.migrate.Apply("task_environments.container_control_auth_token_secret_id", `ALTER TABLE task_environments ADD COLUMN container_control_auth_token_secret_id TEXT DEFAULT ''`) + r.migrate.Apply("task_environments.ownership_generation", `ALTER TABLE task_environments ADD COLUMN ownership_generation INTEGER NOT NULL DEFAULT 1`) if err := r.migrateTaskEnvironmentsRemoveAgentExecutionID(); err != nil { return err } @@ -943,6 +944,7 @@ func (r *Repository) migrateTaskEnvironmentsRemoveAgentExecutionID() error { `CREATE TABLE task_environments_new ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, + ownership_generation INTEGER NOT NULL DEFAULT 1, repository_id TEXT DEFAULT '', executor_type TEXT NOT NULL DEFAULT '', executor_id TEXT DEFAULT '', @@ -964,7 +966,7 @@ func (r *Repository) migrateTaskEnvironmentsRemoveAgentExecutionID() error { FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE )`, `INSERT INTO task_environments_new SELECT - id, task_id, repository_id, executor_type, executor_id, executor_profile_id, + id, task_id, ownership_generation, repository_id, executor_type, executor_id, executor_profile_id, control_port, status, '', worktree_id, worktree_path, worktree_branch, workspace_path, container_id, COALESCE(container_bootstrap_nonce_secret_id, ''), COALESCE(container_control_auth_token_secret_id, ''), sandbox_id, COALESCE(task_dir_name, ''), created_at, updated_at diff --git a/apps/backend/internal/task/repository/sqlite/base_schema.go b/apps/backend/internal/task/repository/sqlite/base_schema.go index 4751f8cdbb..ce69355877 100644 --- a/apps/backend/internal/task/repository/sqlite/base_schema.go +++ b/apps/backend/internal/task/repository/sqlite/base_schema.go @@ -983,6 +983,7 @@ const sessionWorktreeSchemaDDL = ` CREATE TABLE IF NOT EXISTS task_environments ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, + ownership_generation INTEGER NOT NULL DEFAULT 1, executor_type TEXT NOT NULL DEFAULT '', executor_id TEXT DEFAULT '', executor_profile_id TEXT DEFAULT '', diff --git a/apps/backend/internal/task/repository/sqlite/errors.go b/apps/backend/internal/task/repository/sqlite/errors.go index e52ee5e9dc..cbcf8139b0 100644 --- a/apps/backend/internal/task/repository/sqlite/errors.go +++ b/apps/backend/internal/task/repository/sqlite/errors.go @@ -28,6 +28,12 @@ var ErrTaskPlanNotFound = repoerrors.ErrTaskPlanNotFound // the supplied id. Callers should classify it with errors.Is. var ErrTaskEnvironmentNotFound = repoerrors.ErrTaskEnvironmentNotFound +// ErrTaskEnvironmentOwnershipChanged is returned when a guarded transfer's +// expected owner or generation is stale. +var ErrTaskEnvironmentOwnershipChanged = repoerrors.ErrTaskEnvironmentOwnershipChanged + +var errDetachedWorkspaceTransferNotApplicable = errors.New("detached workspace stewardship transfer not applicable") + // ErrNoPrimarySession is returned by GetPrimarySessionByTaskID when the task // has no primary session row. Callers should use errors.Is to distinguish this // "not found" case from genuine backend/DB errors. diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 19d9db8100..05da797516 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "strings" "time" @@ -1234,32 +1235,263 @@ func pendingTaskMetadataMergeExpression(driver string) string { // 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, - ) + tx, err := r.db.BeginTxx(ctx, nil) if err != nil { return false, err } + defer func() { _ = tx.Rollback() }() + parentID, groupID, err := r.loadDetachmentState(ctx, tx, taskID) + if err != nil { + return false, err + } + if parentID == "" { + return false, tx.Commit() + } + lockedTaskIDs := []string{parentID, taskID} + sort.Strings(lockedTaskIDs) + for _, lockedTaskID := range lockedTaskIDs { + if err := r.taskCleanupBarrierLocked(ctx, tx, lockedTaskID); err != nil { + return false, err + } + } + lockedParentID, lockedGroupID, err := r.loadDetachmentState(ctx, tx, taskID) + if err != nil { + return false, err + } + if lockedParentID != parentID || lockedGroupID != groupID { + return false, fmt.Errorf("detach task %s: hierarchy changed concurrently", taskID) + } + + result, err := tx.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 + if rows != 1 { + return false, fmt.Errorf("detach task %s: hierarchy changed concurrently", taskID) } - - 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) + if groupID != "" { + if err := r.transferDetachedWorkspaceStewardship(ctx, tx, groupID, taskID); err != nil { + if !errors.Is(err, errDetachedWorkspaceTransferNotApplicable) { + return false, err + } } - return false, err } - return false, nil + return true, tx.Commit() +} + +func (r *Repository) loadDetachmentState(ctx context.Context, tx *sqlx.Tx, taskID string) (string, string, error) { + var parentID, rawMetadata string + if err := tx.QueryRowContext(ctx, r.db.Rebind( + `SELECT parent_id, COALESCE(metadata, '{}') FROM tasks WHERE id = ?`, + ), taskID).Scan(&parentID, &rawMetadata); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", "", fmt.Errorf("%w: %s", ErrTaskNotFound, taskID) + } + return "", "", err + } + var metadata map[string]interface{} + if trimmed := strings.TrimSpace(rawMetadata); trimmed != "" { + if err := json.Unmarshal([]byte(trimmed), &metadata); err != nil { + return "", "", fmt.Errorf("decode task workspace metadata: %w", err) + } + } + workspace, _ := metadata["workspace"].(map[string]interface{}) + groupID, _ := workspace["group_id"].(string) + if groupID == "" { + if err := tx.QueryRowContext(ctx, r.db.Rebind(` + SELECT workspace_group_id + FROM task_workspace_group_members + WHERE task_id = ? AND released_at IS NULL + ORDER BY created_at DESC, workspace_group_id DESC + LIMIT 1 + `), taskID).Scan(&groupID); err != nil && !errors.Is(err, sql.ErrNoRows) { + // Minimal legacy task databases may not have the optional office + // membership tables. In that case metadata remains the only + // available ownership signal; a missing table is not a detach + // failure. + if !internaldb.IsMissingTableError(err) { + return "", "", err + } + } + } + return parentID, groupID, nil +} + +func (r *Repository) transferDetachedWorkspaceStewardship( + ctx context.Context, + tx *sqlx.Tx, + groupID, taskID string, +) error { + // Read the prospective owners without locking workspace rows first. The + // corresponding task rows must be locked before any group/environment row + // locks, otherwise two concurrent detachments can acquire those resources + // in opposite order and deadlock on PostgreSQL. + state, err := r.loadDetachedWorkspaceStewardship(ctx, tx, groupID, taskID, false) + if err != nil { + return err + } + if err := r.lockDetachedWorkspaceTransferTaskRows(ctx, tx, taskID, state); err != nil { + return err + } + lockedState, err := r.loadDetachedWorkspaceStewardship(ctx, tx, groupID, taskID, true) + if err != nil { + return err + } + if lockedState.groupOwnerTaskID != state.groupOwnerTaskID || + lockedState.environmentOwnerTaskID != state.environmentOwnerTaskID { + // Ownership changed between the unlocked read and the task-row locks. + // Acquire any newly-discovered task barriers before applying the final + // state; all normal paths already hold their initial barriers. + if err := r.lockDetachedWorkspaceTransferTaskRows(ctx, tx, taskID, lockedState); err != nil { + return err + } + lockedState, err = r.loadDetachedWorkspaceStewardship(ctx, tx, groupID, taskID, true) + if err != nil { + return err + } + } + return r.applyDetachedWorkspaceStewardship(ctx, tx, groupID, taskID, lockedState) +} + +type detachedWorkspaceStewardship struct { + groupOwnerTaskID string + environmentID string + environmentOwnerTaskID string + environmentGeneration int64 +} + +func (r *Repository) loadDetachedWorkspaceStewardship( + ctx context.Context, + tx *sqlx.Tx, + groupID, taskID string, + lockRows bool, +) (detachedWorkspaceStewardship, error) { + query := ` + SELECT g.owner_task_id, g.materialized_environment_id + FROM task_workspace_groups g + JOIN task_workspace_group_members m ON m.workspace_group_id = g.id + WHERE g.id = ? AND m.task_id = ? AND m.released_at IS NULL + AND g.cleanup_status IN ('active', 'cleanup_failed') + ` + if lockRows && dialect.IsPostgres(r.db.DriverName()) { + query += ` FOR UPDATE OF g, m` + } + var state detachedWorkspaceStewardship + if err := tx.QueryRowContext(ctx, r.db.Rebind(query), groupID, taskID). + Scan(&state.groupOwnerTaskID, &state.environmentID); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return state, fmt.Errorf("%w: detach task %s: active workspace group %s not found", errDetachedWorkspaceTransferNotApplicable, taskID, groupID) + } + return state, err + } + if state.environmentID == "" { + return state, nil + } + owner, generation, err := r.loadDetachedEnvironmentOwnership(ctx, tx, state.environmentID, taskID, lockRows) + state.environmentOwnerTaskID = owner + state.environmentGeneration = generation + return state, err +} + +func (r *Repository) loadDetachedEnvironmentOwnership( + ctx context.Context, + tx *sqlx.Tx, + environmentID, taskID string, + lockRow bool, +) (string, int64, error) { + query := taskEnvironmentOwnershipQuery + if lockRow && dialect.IsPostgres(r.db.DriverName()) { + query += ` FOR UPDATE` + } + var ownerTaskID string + var generation int64 + if err := tx.QueryRowContext(ctx, r.db.Rebind(query), environmentID). + Scan(&ownerTaskID, &generation); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", 0, fmt.Errorf("detach task %s: materialized environment %s not found", taskID, environmentID) + } + return "", 0, err + } + return ownerTaskID, generation, nil +} + +func (r *Repository) lockDetachedWorkspaceTransferTaskRows( + ctx context.Context, + tx *sqlx.Tx, + taskID string, + state detachedWorkspaceStewardship, +) error { + ids := make([]string, 0, 2) + if state.groupOwnerTaskID != "" && state.groupOwnerTaskID != taskID { + ids = append(ids, state.groupOwnerTaskID) + } + if state.environmentOwnerTaskID != "" && + state.environmentOwnerTaskID != taskID && + state.environmentOwnerTaskID != state.groupOwnerTaskID { + ids = append(ids, state.environmentOwnerTaskID) + } + sort.Strings(ids) + for _, ownerTaskID := range ids { + if err := r.taskCleanupBarrierLocked(ctx, tx, ownerTaskID); err != nil { + return err + } + } + return nil +} + +func (r *Repository) applyDetachedWorkspaceStewardship( + ctx context.Context, + tx *sqlx.Tx, + groupID, taskID string, + state detachedWorkspaceStewardship, +) error { + now := time.Now().UTC() + if _, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE task_workspace_group_members SET role = 'member' + WHERE workspace_group_id = ? AND released_at IS NULL + `), groupID); err != nil { + return err + } + if result, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE task_workspace_group_members SET role = 'owner' + WHERE workspace_group_id = ? AND task_id = ? AND released_at IS NULL + `), groupID, taskID); err != nil { + return err + } else if rows, rowsErr := result.RowsAffected(); rowsErr != nil || rows != 1 { + return errors.Join(rowsErr, fmt.Errorf("detach task %s: owner membership changed concurrently", taskID)) + } + if _, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE task_workspace_groups + SET owner_task_id = ?, ownership_generation = ownership_generation + 1, + cleanup_status = 'active', cleanup_error = '', cleaned_at = NULL, updated_at = ? + WHERE id = ? + `), taskID, now, groupID); err != nil { + return err + } + if state.environmentID == "" { + return nil + } + result, err := tx.ExecContext(ctx, r.db.Rebind(` + UPDATE task_environments + SET task_id = ?, ownership_generation = ownership_generation + 1, updated_at = ? + WHERE id = ? AND task_id = ? AND ownership_generation = ? + `), taskID, now, state.environmentID, state.environmentOwnerTaskID, state.environmentGeneration) + if err != nil { + return err + } + rows, err := result.RowsAffected() + if err != nil { + return err + } + if rows != 1 { + return fmt.Errorf("detach task %s: materialized environment %s changed concurrently", taskID, state.environmentID) + } + return nil } func detachTaskQuery(driver string) string { diff --git a/apps/backend/internal/task/repository/sqlite/task_environment.go b/apps/backend/internal/task/repository/sqlite/task_environment.go index 8302d852ae..ec34182dae 100644 --- a/apps/backend/internal/task/repository/sqlite/task_environment.go +++ b/apps/backend/internal/task/repository/sqlite/task_environment.go @@ -3,15 +3,19 @@ package sqlite import ( "context" "database/sql" + "errors" "fmt" "time" "github.com/google/uuid" "github.com/jmoiron/sqlx" + "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/task/models" ) +const taskEnvironmentOwnershipQuery = `SELECT task_id, ownership_generation FROM task_environments WHERE id = ?` + // CreateTaskEnvironment creates a new task environment record. // If env.Repos is non-empty, the per-repo rows are inserted in the same transaction. func (r *Repository) CreateTaskEnvironment(ctx context.Context, env *models.TaskEnvironment) error { @@ -30,6 +34,9 @@ func (r *Repository) CreateTaskEnvironment(ctx context.Context, env *models.Task now := time.Now().UTC() env.CreatedAt = now env.UpdatedAt = now + if env.OwnershipGeneration == 0 { + env.OwnershipGeneration = 1 + } tx, err := r.db.BeginTxx(ctx, nil) if err != nil { @@ -60,14 +67,14 @@ func (r *Repository) CreateTaskEnvironment(ctx context.Context, env *models.Task if _, err := tx.ExecContext(ctx, r.db.Rebind(` INSERT INTO task_environments ( - id, task_id, executor_type, executor_id, executor_profile_id, + id, task_id, ownership_generation, executor_type, executor_id, executor_profile_id, control_port, status, materialization_session_id, workspace_path, container_id, container_bootstrap_nonce_secret_id, container_control_auth_token_secret_id, sandbox_id, task_dir_name, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `), - env.ID, env.TaskID, env.ExecutorType, env.ExecutorID, env.ExecutorProfileID, + env.ID, env.TaskID, env.OwnershipGeneration, env.ExecutorType, env.ExecutorID, env.ExecutorProfileID, env.ControlPort, string(env.Status), env.MaterializationSessionID, env.WorkspacePath, env.ContainerID, env.ContainerBootstrapNonceSecretID, env.ContainerControlAuthTokenSecretID, env.SandboxID, env.TaskDirName, @@ -92,14 +99,14 @@ func (r *Repository) GetTaskEnvironment(ctx context.Context, id string) (*models var status string err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` - SELECT id, task_id, executor_type, executor_id, executor_profile_id, + SELECT id, task_id, ownership_generation, executor_type, executor_id, executor_profile_id, control_port, status, materialization_session_id, workspace_path, container_id, COALESCE(container_bootstrap_nonce_secret_id, ''), COALESCE(container_control_auth_token_secret_id, ''), sandbox_id, COALESCE(task_dir_name, ''), created_at, updated_at FROM task_environments WHERE id = ? `), id).Scan( - &env.ID, &env.TaskID, &env.ExecutorType, &env.ExecutorID, &env.ExecutorProfileID, + &env.ID, &env.TaskID, &env.OwnershipGeneration, &env.ExecutorType, &env.ExecutorID, &env.ExecutorProfileID, &env.ControlPort, &status, &env.MaterializationSessionID, &env.WorkspacePath, &env.ContainerID, &env.ContainerBootstrapNonceSecretID, &env.ContainerControlAuthTokenSecretID, &env.SandboxID, &env.TaskDirName, @@ -128,14 +135,14 @@ func (r *Repository) GetTaskEnvironmentByTaskID(ctx context.Context, taskID stri var status string err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` - SELECT id, task_id, executor_type, executor_id, executor_profile_id, + SELECT id, task_id, ownership_generation, executor_type, executor_id, executor_profile_id, control_port, status, materialization_session_id, workspace_path, container_id, COALESCE(container_bootstrap_nonce_secret_id, ''), COALESCE(container_control_auth_token_secret_id, ''), sandbox_id, COALESCE(task_dir_name, ''), created_at, updated_at FROM task_environments WHERE task_id = ? ORDER BY created_at DESC LIMIT 1 `), taskID).Scan( - &env.ID, &env.TaskID, &env.ExecutorType, &env.ExecutorID, &env.ExecutorProfileID, + &env.ID, &env.TaskID, &env.OwnershipGeneration, &env.ExecutorType, &env.ExecutorID, &env.ExecutorProfileID, &env.ControlPort, &status, &env.MaterializationSessionID, &env.WorkspacePath, &env.ContainerID, &env.ContainerBootstrapNonceSecretID, &env.ContainerControlAuthTokenSecretID, &env.SandboxID, &env.TaskDirName, @@ -590,10 +597,54 @@ func (r *Repository) taskHasRepositoriesTx(ctx context.Context, tx *sqlx.Tx, tas return count > 0, nil } -func (r *Repository) TransferTaskEnvironmentToTask(ctx context.Context, envID, taskID string) error { - result, err := r.db.ExecContext(ctx, r.db.Rebind(` +// TransferTaskEnvironmentOwnership moves an environment only when the +// caller's owner-generation snapshot is still current. +func (r *Repository) TransferTaskEnvironmentOwnership( + ctx context.Context, + envID, expectedTaskID string, + expectedGeneration int64, + taskID string, +) error { + return r.transferTaskEnvironmentOwnership(ctx, envID, expectedTaskID, expectedGeneration, taskID) +} + +func (r *Repository) transferTaskEnvironmentOwnership( + ctx context.Context, + envID, expectedTaskID string, + expectedGeneration int64, + taskID string, +) error { + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + query := taskEnvironmentOwnershipQuery + if dialect.IsPostgres(r.db.DriverName()) { + query += ` FOR UPDATE` + } + var currentTaskID string + var currentGeneration int64 + if err := tx.QueryRowContext(ctx, r.db.Rebind(query), envID).Scan(¤tTaskID, ¤tGeneration); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("%w: %s", ErrTaskEnvironmentNotFound, envID) + } + return err + } + if currentTaskID != expectedTaskID || currentGeneration != expectedGeneration { + return fmt.Errorf("%w: environment %s is owned by %s at generation %d", + ErrTaskEnvironmentOwnershipChanged, envID, currentTaskID, currentGeneration) + } + if currentTaskID == taskID { + return tx.Commit() + } + if err := r.taskCleanupBarrierLocked(ctx, tx, currentTaskID); err != nil { + return err + } + result, err := tx.ExecContext(ctx, r.db.Rebind(` UPDATE task_environments SET task_id = ?, + ownership_generation = ownership_generation + 1, updated_at = ? WHERE id = ? `), taskID, time.Now().UTC(), envID) @@ -604,7 +655,74 @@ func (r *Repository) TransferTaskEnvironmentToTask(ctx context.Context, envID, t if rows == 0 { return fmt.Errorf("%w: %s", ErrTaskEnvironmentNotFound, envID) } - return nil + return tx.Commit() +} + +// ClaimTaskEnvironmentReset reserves destructive environment reset behind the +// same task cleanup barrier used by lifecycle cleanup. The prepared barrier +// remains in place while external resources are torn down, so ownership +// transfers cannot race the reset's snapshot. +func (r *Repository) ClaimTaskEnvironmentReset( + ctx context.Context, + taskID, environmentID string, + expectedGeneration int64, + operationID string, +) (string, error) { + if taskID == "" || environmentID == "" || operationID == "" { + return "", errors.New("task environment reset claim: task, environment, and operation ids are required") + } + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return "", err + } + defer func() { _ = tx.Rollback() }() + if err := r.lockTaskRowInTx(ctx, tx, taskID); err != nil { + return "", err + } + if err := r.taskCleanupBarrierLocked(ctx, tx, taskID); err != nil { + return "", err + } + now := time.Now().UTC() + jobID := uuid.NewString() + if _, err := tx.ExecContext(ctx, r.db.Rebind(` + INSERT INTO task_resource_cleanup_jobs ( + id, operation_id, task_id, trigger, state, resource_snapshot, attempts, + next_attempt_at, last_error, created_at, updated_at, completed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `), jobID, operationID, taskID, models.TaskResourceCleanupTriggerReconcile, + models.TaskResourceCleanupStatePrepared, `{}`, 0, nil, "", now, now, nil); err != nil { + return "", err + } + query := taskEnvironmentOwnershipQuery + if dialect.IsPostgres(r.db.DriverName()) { + query += ` FOR UPDATE` + } + var ownerTaskID string + var generation int64 + if err := tx.QueryRowContext(ctx, r.db.Rebind(query), environmentID).Scan(&ownerTaskID, &generation); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", fmt.Errorf("%w: %s", ErrTaskEnvironmentNotFound, environmentID) + } + return "", err + } + if ownerTaskID != taskID || generation != expectedGeneration { + return "", fmt.Errorf("%w: environment %s is owned by %s at generation %d", + ErrTaskEnvironmentOwnershipChanged, environmentID, ownerTaskID, generation) + } + if err := tx.Commit(); err != nil { + return "", err + } + return jobID, nil +} + +// ReleaseTaskEnvironmentReset removes a reset's durable cleanup barrier. +func (r *Repository) ReleaseTaskEnvironmentReset( + ctx context.Context, + jobID string, + state models.TaskResourceCleanupState, + errStr string, +) error { + return r.CompleteTaskResourceCleanupJob(ctx, jobID, state, errStr, nil) } // DeleteTaskEnvironment deletes a task environment by ID. diff --git a/apps/backend/internal/task/repository/sqlite/task_environment_test.go b/apps/backend/internal/task/repository/sqlite/task_environment_test.go index 2f8d4f29b2..e44c6f6dd3 100644 --- a/apps/backend/internal/task/repository/sqlite/task_environment_test.go +++ b/apps/backend/internal/task/repository/sqlite/task_environment_test.go @@ -102,6 +102,86 @@ func TestGetTaskEnvironmentMissingReturnsSentinel(t *testing.T) { } } +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.4 +func TestTransferTaskEnvironmentAdvancesGenerationAndHonorsCleanupBarrier(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedWorkspace(t, repo, "workspace-environment-transfer") + for _, taskID := range []string{"owner-a", "owner-b", "owner-c"} { + if err := repo.CreateTask(ctx, &models.Task{ID: taskID, WorkspaceID: "workspace-environment-transfer", Title: taskID}); err != nil { + t.Fatalf("CreateTask(%s): %v", taskID, err) + } + } + if err := repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ + ID: "environment-transfer", TaskID: "owner-a", ExecutorType: string(models.ExecutorTypeLocal), + Status: models.TaskEnvironmentStatusReady, + }); err != nil { + t.Fatalf("CreateTaskEnvironment: %v", err) + } + + if err := repo.TransferTaskEnvironmentOwnership(ctx, "environment-transfer", "owner-a", 1, "owner-b"); err != nil { + t.Fatalf("TransferTaskEnvironmentOwnership: %v", err) + } + transferred, err := repo.GetTaskEnvironment(ctx, "environment-transfer") + if err != nil { + t.Fatalf("GetTaskEnvironment: %v", err) + } + if transferred.TaskID != "owner-b" || transferred.OwnershipGeneration != 2 { + t.Fatalf("transferred environment = owner %q generation %d; want owner-b, 2", transferred.TaskID, transferred.OwnershipGeneration) + } + if err := repo.TransferTaskEnvironmentOwnership(ctx, "environment-transfer", "owner-a", 1, "owner-c"); err == nil { + t.Fatal("stale ownership retry transferred the current generation") + } + + if err := repo.CreateTaskResourceCleanupJob(ctx, &models.TaskResourceCleanupJob{ + ID: "cleanup-owner-b", OperationID: "delete:owner-b", TaskID: "owner-b", + Trigger: models.TaskResourceCleanupTriggerDelete, State: models.TaskResourceCleanupStatePending, + }); err != nil { + t.Fatalf("CreateTaskResourceCleanupJob: %v", err) + } + if err := repo.TransferTaskEnvironmentOwnership(ctx, "environment-transfer", "owner-b", 2, "owner-c"); err == nil { + t.Fatal("TransferTaskEnvironmentOwnership succeeded through active cleanup barrier") + } + current, err := repo.GetTaskEnvironment(ctx, "environment-transfer") + if err != nil { + t.Fatalf("GetTaskEnvironment after rejected transfer: %v", err) + } + if current.TaskID != "owner-b" || current.OwnershipGeneration != 2 { + t.Fatalf("environment after rejected transfer = owner %q generation %d; want owner-b, 2", current.TaskID, current.OwnershipGeneration) + } +} + +func TestClaimTaskEnvironmentResetBlocksOwnershipTransferUntilReleased(t *testing.T) { + repo := newRepoForEntityTests(t) + ctx := context.Background() + seedWorkspace(t, repo, "workspace-environment-reset-claim") + for _, taskID := range []string{"reset-owner", "reset-borrower"} { + if err := repo.CreateTask(ctx, &models.Task{ID: taskID, WorkspaceID: "workspace-environment-reset-claim", Title: taskID}); err != nil { + t.Fatalf("CreateTask(%s): %v", taskID, err) + } + } + if err := repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ + ID: "environment-reset-claim", TaskID: "reset-owner", ExecutorType: string(models.ExecutorTypeLocal), + Status: models.TaskEnvironmentStatusReady, + }); err != nil { + t.Fatalf("CreateTaskEnvironment: %v", err) + } + + jobID, err := repo.ClaimTaskEnvironmentReset(ctx, "reset-owner", "environment-reset-claim", 1, "reset:operation") + if err != nil { + t.Fatalf("ClaimTaskEnvironmentReset: %v", err) + } + if err := repo.TransferTaskEnvironmentOwnership(ctx, "environment-reset-claim", "reset-owner", 1, "reset-borrower"); err == nil { + t.Fatal("TransferTaskEnvironmentOwnership succeeded while reset claim was active") + } + if err := repo.ReleaseTaskEnvironmentReset(ctx, jobID, models.TaskResourceCleanupStateSucceeded, ""); err != nil { + t.Fatalf("ReleaseTaskEnvironmentReset: %v", err) + } + if err := repo.TransferTaskEnvironmentOwnership(ctx, "environment-reset-claim", "reset-owner", 1, "reset-borrower"); err != nil { + t.Fatalf("TransferTaskEnvironmentOwnership after release: %v", err) + } +} + func TestTaskEnvironment_PersistsDockerBootstrapNonceReference(t *testing.T) { repo := newRepoForEntityTests(t) ctx := context.Background() @@ -632,10 +712,10 @@ func TestUpdateTaskEnvironmentAllowsFailedMaterializationWithoutWorkspacePath(t func TestTransferTaskEnvironmentMissingReturnsSentinel(t *testing.T) { repo := newRepoForHealTests(t) - err := repo.TransferTaskEnvironmentToTask(context.Background(), "missing-environment", "task-1") + err := repo.TransferTaskEnvironmentOwnership(context.Background(), "missing-environment", "task-1", 1, "task-2") if !errors.Is(err, ErrTaskEnvironmentNotFound) { - t.Fatalf("TransferTaskEnvironmentToTask error = %v, want ErrTaskEnvironmentNotFound", err) + t.Fatalf("TransferTaskEnvironmentOwnership error = %v, want ErrTaskEnvironmentNotFound", err) } } diff --git a/apps/backend/internal/task/repository/sqlite/worktree_ownership_migration.go b/apps/backend/internal/task/repository/sqlite/worktree_ownership_migration.go index 8f6b27e693..c8ebe49a42 100644 --- a/apps/backend/internal/task/repository/sqlite/worktree_ownership_migration.go +++ b/apps/backend/internal/task/repository/sqlite/worktree_ownership_migration.go @@ -38,6 +38,7 @@ func finalTaskEnvironmentsDDL(tableName string) string { CREATE TABLE IF NOT EXISTS %s ( id TEXT PRIMARY KEY, task_id TEXT NOT NULL, + ownership_generation INTEGER NOT NULL DEFAULT 1, executor_type TEXT NOT NULL DEFAULT '', executor_id TEXT DEFAULT '', executor_profile_id TEXT DEFAULT '', diff --git a/apps/backend/internal/task/repository/sqlite/worktree_ownership_targets.go b/apps/backend/internal/task/repository/sqlite/worktree_ownership_targets.go index 2ebed8777b..ebe579d7fc 100644 --- a/apps/backend/internal/task/repository/sqlite/worktree_ownership_targets.go +++ b/apps/backend/internal/task/repository/sqlite/worktree_ownership_targets.go @@ -239,11 +239,11 @@ func (r *Repository) insertNormalized(c *worktreeCutover, tx *sqlx.Tx) error { } if _, err := tx.Exec(tx.Rebind(` INSERT INTO task_environments_shadow ( - id, task_id, executor_type, executor_id, executor_profile_id, + id, task_id, ownership_generation, executor_type, executor_id, executor_profile_id, control_port, status, materialization_session_id, workspace_path, container_id, container_bootstrap_nonce_secret_id, container_control_auth_token_secret_id, sandbox_id, task_dir_name, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), - env.id, taskID, env.executorType, env.executorID, env.executorProfileID, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), + env.id, taskID, int64(1), env.executorType, env.executorID, env.executorProfileID, env.controlPort, env.status, "", env.workspacePath, env.containerID, env.containerBootstrapNonceSecretID, env.containerControlAuthTokenSecretID, env.sandboxID, env.taskDirName, env.createdAt, env.updatedAt); err != nil { @@ -436,7 +436,7 @@ func (r *Repository) checkShadowFinalSchema(tx *sqlx.Tx) error { return fmt.Errorf("cutover: task_environments_shadow still carries legacy column %s", legacy) } } - for _, required := range []string{"id", "task_id", "executor_type", columnStatus, "workspace_path", columnCreatedAt, columnUpdatedAt} { + for _, required := range []string{"id", "task_id", "ownership_generation", "executor_type", columnStatus, "workspace_path", columnCreatedAt, columnUpdatedAt} { if !envColumns[required] { return fmt.Errorf("cutover: task_environments_shadow missing final column %s", required) } diff --git a/apps/backend/internal/task/service/create_task_references.go b/apps/backend/internal/task/service/create_task_references.go index 95177ba974..b92ccc121c 100644 --- a/apps/backend/internal/task/service/create_task_references.go +++ b/apps/backend/internal/task/service/create_task_references.go @@ -119,6 +119,12 @@ func (s *Service) rollbackPartialTask(ctx context.Context, taskID string, cause s.logger.Error("rollback delete failed; task left in inconsistent state", zap.String("task_id", taskID), zap.Error(err)) } + if releaser, ok := s.workspacePolicyAttacher.(WorkspacePolicyMembershipReleaser); ok { + if err := releaser.ReleaseWorkspacePolicy(rollbackCtx, taskID, "create_rollback"); err != nil { + s.logger.Warn("rollback workspace membership cleanup failed", + zap.String("task_id", taskID), zap.Error(err)) + } + } return cause } diff --git a/apps/backend/internal/task/service/handoff_cascade.go b/apps/backend/internal/task/service/handoff_cascade.go index 565d1d08d2..8a0f2928ea 100644 --- a/apps/backend/internal/task/service/handoff_cascade.go +++ b/apps/backend/internal/task/service/handoff_cascade.go @@ -22,10 +22,11 @@ type workspaceEnvironmentRepository interface { } type workspaceEnvironmentOwnershipTransfer struct { - groupID string - environmentID string - oldOwnerTaskID string - newOwnerTaskID string + groupID string + environmentID string + oldOwnerTaskID string + newOwnerTaskID string + resultingGeneration int64 } // publishUpdatedTask re-reads the task row and forwards it to the event @@ -95,12 +96,15 @@ func (s *HandoffService) evaluateWorkspaceGroupCleanup(ctx context.Context, grou orchmodels.WorkspaceCleanupStatusPending, "active executor still bound to group's member session", nil) } - // Mark the group as cleanup_pending before invoking the cleaner so - // any concurrent observer sees a consistent state machine. - if err := s.wsGroups.UpdateWorkspaceGroupCleanupStatus(ctx, groupID, - orchmodels.WorkspaceCleanupStatusPending, "", nil); err != nil { + // Claim this ownership generation before invoking the cleaner. A stale + // evaluator or a concurrent membership admission cannot pass this write. + claimed, err := claimWorkspaceGroupCleanup(ctx, s.wsGroups, g) + if err != nil { return err } + if !claimed { + return nil + } if s.cleaner == nil { // No cleaner wired — leave the group in cleanup_pending so the // operator can see it awaiting a cleaner upgrade. This matches @@ -108,12 +112,12 @@ func (s *HandoffService) evaluateWorkspaceGroupCleanup(ctx context.Context, grou return nil } if err := s.runWorkspaceGroupCleanup(ctx, g); err != nil { - _ = s.wsGroups.UpdateWorkspaceGroupCleanupStatus(ctx, groupID, + _ = completeWorkspaceGroupCleanup(ctx, s.wsGroups, g, orchmodels.WorkspaceCleanupStatusFailed, err.Error(), nil) return err } now := time.Now().UTC() - return s.wsGroups.UpdateWorkspaceGroupCleanupStatus(ctx, groupID, + return completeWorkspaceGroupCleanup(ctx, s.wsGroups, g, orchmodels.WorkspaceCleanupStatusCleaned, "", &now) } @@ -527,7 +531,9 @@ func (s *HandoffService) transferWorkspaceGroupEnvironmentOwnership( if newOwner == "" { return nil, fmt.Errorf("preserve workspace group %s: no surviving member can own environment %s", group.ID, env.ID) } - if err := environments.TransferTaskEnvironmentToTask(ctx, env.ID, newOwner); err != nil { + if err := environments.TransferTaskEnvironmentOwnership( + ctx, env.ID, env.TaskID, env.OwnershipGeneration, newOwner, + ); err != nil { return nil, fmt.Errorf("transfer workspace environment %s to task %s: %w", env.ID, newOwner, err) } s.logf().Info("transferred shared workspace environment before task lifecycle cleanup", @@ -536,10 +542,11 @@ func (s *HandoffService) transferWorkspaceGroupEnvironmentOwnership( zap.String("old_owner_task_id", env.TaskID), zap.String("new_owner_task_id", newOwner)) return &workspaceEnvironmentOwnershipTransfer{ - groupID: group.ID, - environmentID: env.ID, - oldOwnerTaskID: env.TaskID, - newOwnerTaskID: newOwner, + groupID: group.ID, + environmentID: env.ID, + oldOwnerTaskID: env.TaskID, + newOwnerTaskID: newOwner, + resultingGeneration: env.OwnershipGeneration + 1, }, nil } @@ -580,8 +587,12 @@ func (s *HandoffService) rollbackWorkspaceEnvironmentOwnershipTransfers( // A concurrent retry already restored this transfer. case env.TaskID != transfer.newOwnerTaskID: err = fmt.Errorf("owner changed from rollback target %s to %s", transfer.newOwnerTaskID, env.TaskID) + case env.OwnershipGeneration != transfer.resultingGeneration: + err = fmt.Errorf("ownership generation changed from rollback target %d to %d", transfer.resultingGeneration, env.OwnershipGeneration) default: - err = environments.TransferTaskEnvironmentToTask(ctx, transfer.environmentID, transfer.oldOwnerTaskID) + err = environments.TransferTaskEnvironmentOwnership( + ctx, transfer.environmentID, transfer.newOwnerTaskID, transfer.resultingGeneration, transfer.oldOwnerTaskID, + ) } } mu.Unlock() diff --git a/apps/backend/internal/task/service/handoff_cleaner.go b/apps/backend/internal/task/service/handoff_cleaner.go index 8aba25c453..6ccc0bfa2f 100644 --- a/apps/backend/internal/task/service/handoff_cleaner.go +++ b/apps/backend/internal/task/service/handoff_cleaner.go @@ -62,26 +62,24 @@ func (s *HandoffService) CleanupWorkspaceGroups(ctx context.Context, workspaceID s.logf().Warn("workspace group cleanup: active executions remain", zap.String("workspace_id", workspaceID), zap.String("group_id", g.ID)) - if err := s.wsGroups.UpdateWorkspaceGroupCleanupStatus(statusCtx, g.ID, - orchmodels.WorkspaceCleanupStatusPending, - "active executor still bound during workspace deletion", nil); err != nil { - return err - } if err := s.waitForWorkspaceGroupIdle(ctx, workspaceID, g.ID); err != nil { return err } } - if err := s.wsGroups.UpdateWorkspaceGroupCleanupStatus(statusCtx, g.ID, - orchmodels.WorkspaceCleanupStatusPending, "", nil); err != nil { + claimed, err := claimWorkspaceGroupCleanup(statusCtx, s.wsGroups, g) + if err != nil { return err } + if !claimed { + continue + } if err := s.runWorkspaceGroupCleanup(ctx, g); err != nil { - _ = s.wsGroups.UpdateWorkspaceGroupCleanupStatus(statusCtx, g.ID, + _ = completeWorkspaceGroupCleanup(statusCtx, s.wsGroups, g, orchmodels.WorkspaceCleanupStatusFailed, err.Error(), nil) return fmt.Errorf("clean workspace group %s: %w", g.ID, err) } now := time.Now().UTC() - if err := s.wsGroups.UpdateWorkspaceGroupCleanupStatus(statusCtx, g.ID, + if err := completeWorkspaceGroupCleanup(statusCtx, s.wsGroups, g, orchmodels.WorkspaceCleanupStatusCleaned, "", &now); err != nil { return err } diff --git a/apps/backend/internal/task/service/handoff_service.go b/apps/backend/internal/task/service/handoff_service.go index d5981b03cf..b3dfa29a27 100644 --- a/apps/backend/internal/task/service/handoff_service.go +++ b/apps/backend/internal/task/service/handoff_service.go @@ -21,6 +21,7 @@ import ( // in sync with internal/mcp/handlers/workspace_policy.go. const ( workspaceModeInheritParent = "inherit_parent" + workspaceModeNewWorkspace = "new_workspace" workspaceModeSharedGroup = "shared_group" orderingSequential = "sequential" ) @@ -51,6 +52,47 @@ type WorkspaceGroupRepo interface { UpdateWorkspaceGroupRestoreStatus(ctx context.Context, id, status, errStr string) error } +type workspaceGroupCleanupGenerationRepository interface { + ClaimWorkspaceGroupCleanup(ctx context.Context, id string, generation int64) (bool, error) + CompleteWorkspaceGroupCleanup( + ctx context.Context, + id string, + generation int64, + status, errStr string, + cleanedAt *time.Time, + ) (bool, error) +} + +func claimWorkspaceGroupCleanup( + ctx context.Context, + repo WorkspaceGroupRepo, + group *orchmodels.WorkspaceGroup, +) (bool, error) { + if guarded, ok := repo.(workspaceGroupCleanupGenerationRepository); ok { + return guarded.ClaimWorkspaceGroupCleanup(ctx, group.ID, group.OwnershipGeneration) + } + err := repo.UpdateWorkspaceGroupCleanupStatus( + ctx, group.ID, orchmodels.WorkspaceCleanupStatusPending, "", nil, + ) + return err == nil, err +} + +func completeWorkspaceGroupCleanup( + ctx context.Context, + repo WorkspaceGroupRepo, + group *orchmodels.WorkspaceGroup, + status, errStr string, + cleanedAt *time.Time, +) error { + if guarded, ok := repo.(workspaceGroupCleanupGenerationRepository); ok { + _, err := guarded.CompleteWorkspaceGroupCleanup( + ctx, group.ID, group.OwnershipGeneration, status, errStr, cleanedAt, + ) + return err + } + return repo.UpdateWorkspaceGroupCleanupStatus(ctx, group.ID, status, errStr, cleanedAt) +} + // SessionWorktreeReader is the slim repository surface HandoffService // uses to discover a task's primary session + worktree details when // recording materialization state on a workspace group. Implemented by @@ -406,15 +448,33 @@ func (s *HandoffService) AttachWorkspacePolicy(ctx context.Context, taskID, pare if !pol.NeedsAttachment() { return nil } - if err := s.attachWorkspaceGroup(ctx, taskID, parentID, pol); err != nil { - return err - } if pol.ParentOrdering == orderingSequential && parentID != "" { if err := s.attachSequentialBlocker(ctx, taskID, parentID); err != nil { return err } } - return nil + return s.attachWorkspaceGroup(ctx, taskID, parentID, pol) +} + +// ReleaseWorkspacePolicy removes the task's active workspace-group +// membership after a create rollback. The membership audit row is retained, +// and normal group cleanup evaluation runs so a group created solely for the +// failed task cannot remain active forever. +func (s *HandoffService) ReleaseWorkspacePolicy(ctx context.Context, taskID, reason string) error { + if s.wsGroups == nil || taskID == "" { + return nil + } + group, err := s.wsGroups.GetWorkspaceGroupForTask(ctx, taskID) + if err != nil { + return err + } + if group == nil { + return nil + } + if err := s.wsGroups.ReleaseWorkspaceGroupMember(ctx, group.ID, taskID, reason, ""); err != nil { + return err + } + return s.evaluateWorkspaceGroupCleanup(ctx, group.ID) } func (s *HandoffService) attachWorkspaceGroup(ctx context.Context, taskID, parentID string, pol WorkspacePolicy) error { diff --git a/apps/backend/internal/task/service/handoff_workspace_test.go b/apps/backend/internal/task/service/handoff_workspace_test.go index 1495a47866..f51ee61ba5 100644 --- a/apps/backend/internal/task/service/handoff_workspace_test.go +++ b/apps/backend/internal/task/service/handoff_workspace_test.go @@ -8,6 +8,7 @@ import ( orchmodels "github.com/kandev/kandev/internal/office/models" "github.com/kandev/kandev/internal/task/models" + taskrepo "github.com/kandev/kandev/internal/task/repository" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -346,12 +347,23 @@ func (f *fakeTaskRepo) GetTaskEnvironmentByTaskID(_ context.Context, taskID stri return nil, nil } -func (f *fakeTaskRepo) TransferTaskEnvironmentToTask(_ context.Context, envID, taskID string) error { +func (f *fakeTaskRepo) TransferTaskEnvironmentOwnership( + _ context.Context, + envID, expectedTaskID string, + expectedGeneration int64, + taskID string, +) error { f.mu.Lock() defer f.mu.Unlock() - if env := f.taskEnvironments[envID]; env != nil { - env.TaskID = taskID + env := f.taskEnvironments[envID] + if env == nil { + return nil } + if env.TaskID != expectedTaskID || env.OwnershipGeneration != expectedGeneration { + return taskrepo.ErrTaskEnvironmentOwnershipChanged + } + env.TaskID = taskID + env.OwnershipGeneration++ return nil } @@ -410,8 +422,13 @@ func (r *phase4TaskRepo) GetTaskEnvironment(ctx context.Context, id string) (*mo func (r *phase4TaskRepo) GetTaskEnvironmentByTaskID(ctx context.Context, taskID string) (*models.TaskEnvironment, error) { return r.base.GetTaskEnvironmentByTaskID(ctx, taskID) } -func (r *phase4TaskRepo) TransferTaskEnvironmentToTask(ctx context.Context, envID, taskID string) error { - return r.base.TransferTaskEnvironmentToTask(ctx, envID, taskID) +func (r *phase4TaskRepo) TransferTaskEnvironmentOwnership( + ctx context.Context, + envID, expectedTaskID string, + expectedGeneration int64, + taskID string, +) error { + return r.base.TransferTaskEnvironmentOwnership(ctx, envID, expectedTaskID, expectedGeneration, taskID) } // All other TaskRepository methods panic — the AttachWorkspacePolicy diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 7a68da6ab1..26c5178ea1 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -341,6 +341,7 @@ type Service struct { taskActivity repository.TaskActivityRepository subagentContexts repository.SubagentContextRepository usage repository.UsageRepository + workspacePolicyAttacher WorkspacePolicyAttacher attachmentSvc *AttachmentService statusSummaryPRs TaskStatusSummaryPRReader statusSummaryProjector TaskStatusSummaryEventProjector @@ -427,6 +428,20 @@ type Service struct { lastPendingActionProjections map[string]pendingActionProjectionState } +// WorkspacePolicyAttacher persists the workspace-group relationship that is +// required before a newly created child can be returned or launched. +type WorkspacePolicyAttacher interface { + AttachWorkspacePolicy(ctx context.Context, taskID, parentID string, policy WorkspacePolicy) error +} + +// WorkspacePolicyMembershipReleaser removes a task's workspace-group +// membership after a post-create rollback. It is an optional companion to +// WorkspacePolicyAttacher because lightweight task-service test harnesses may +// not persist workspace groups. +type WorkspacePolicyMembershipReleaser interface { + ReleaseWorkspacePolicy(ctx context.Context, taskID, reason string) error +} + // SetAttachmentService wires the file-backed prompt attachment owner into the // task service. It is optional for focused unit-test harnesses that never send // file-backed descriptors. @@ -459,6 +474,12 @@ func (s *Service) SetWorkspaceSecretDeleter(deleter WorkspaceSecretDeleter) { s.workspaceSecretDeleter = deleter } +// SetWorkspacePolicyAttacher installs the canonical child-workspace +// coordinator used by every CreateTask caller. +func (s *Service) SetWorkspacePolicyAttacher(attacher WorkspacePolicyAttacher) { + s.workspacePolicyAttacher = attacher +} + // NewService creates a new task service func NewService(repos Repos, eventBus bus.EventBus, log *logger.Logger, discoveryConfig RepositoryDiscoveryConfig) *Service { return &Service{ diff --git a/apps/backend/internal/task/service/service_child_task.go b/apps/backend/internal/task/service/service_child_task.go index aee5d9dcc4..55d713b017 100644 --- a/apps/backend/internal/task/service/service_child_task.go +++ b/apps/backend/internal/task/service/service_child_task.go @@ -59,6 +59,7 @@ func (s *Service) CreateChildTask( AssigneeAgentProfileID: assignee, Origin: models.TaskOriginAgentCreated, ProjectID: parent.ProjectID, + WorkspacePolicy: &WorkspacePolicy{Mode: workspaceModeInheritParent}, } result, err := s.CreateTask(ctx, req) if err != nil { diff --git a/apps/backend/internal/task/service/service_child_task_test.go b/apps/backend/internal/task/service/service_child_task_test.go index 2d65ebba70..09f0518c85 100644 --- a/apps/backend/internal/task/service/service_child_task_test.go +++ b/apps/backend/internal/task/service/service_child_task_test.go @@ -6,9 +6,113 @@ import ( "strings" "testing" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" ) +type failingWorkspacePolicyAttacher struct { + err error +} + +func (f failingWorkspacePolicyAttacher) AttachWorkspacePolicy(context.Context, string, string, WorkspacePolicy) error { + return f.err +} + +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.1 +func TestCreateChildTaskAttachesInheritedWorkspaceBeforeReturning(t *testing.T) { + svc, database := createOfficeIntegrationServiceWithDB(t) + ctx := context.Background() + officeRepo, err := officesqlite.NewWithDB(database, database, nil) + if err != nil { + t.Fatalf("office repository: %v", err) + } + taskRepo, ok := svc.tasks.(*sqliterepo.Repository) + if !ok { + t.Fatalf("task repository type = %T", svc.tasks) + } + handoff := NewHandoffService(taskRepo, nil, nil, officeRepo, officeRepo, nil) + svc.SetWorkspacePolicyAttacher(handoff) + + if err := svc.workspaces.CreateWorkspace(ctx, &models.Workspace{ID: "workspace-child-attachment", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + if err := svc.workflows.CreateWorkflow(ctx, &models.Workflow{ID: "workflow-child-attachment", WorkspaceID: "workspace-child-attachment", Name: "Workflow"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + parentResult, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "workspace-child-attachment", + WorkflowID: "workflow-child-attachment", + Title: "Parent", + }) + if err != nil { + t.Fatalf("create parent: %v", err) + } + + childID, err := svc.CreateChildTask(ctx, parentResult.Task, ChildTaskSpec{Title: "Child"}) + if err != nil { + t.Fatalf("CreateChildTask: %v", err) + } + parentGroup, err := officeRepo.GetWorkspaceGroupForTask(ctx, parentResult.Task.ID) + if err != nil { + t.Fatalf("GetWorkspaceGroupForTask(parent): %v", err) + } + childGroup, err := officeRepo.GetWorkspaceGroupForTask(ctx, childID) + if err != nil { + t.Fatalf("GetWorkspaceGroupForTask(child): %v", err) + } + if parentGroup == nil || childGroup == nil || parentGroup.ID != childGroup.ID { + t.Fatalf("workspace groups = parent %#v, child %#v; want one shared group", parentGroup, childGroup) + } +} + +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.2 +func TestCreateChildTaskRollsBackWhenWorkspaceAttachmentFails(t *testing.T) { + svc, repo := setupOfficeTest(t) + ctx := context.Background() + parentResult, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "ws-1", Title: "Parent", ProjectID: "proj-1", + }) + if err != nil { + t.Fatalf("create parent: %v", err) + } + svc.SetWorkspacePolicyAttacher(failingWorkspacePolicyAttacher{err: errors.New("attachment unavailable")}) + + if _, err := svc.CreateChildTask(ctx, parentResult.Task, ChildTaskSpec{Title: "Child"}); err == nil { + t.Fatal("CreateChildTask succeeded after required workspace attachment failed") + } + children, err := repo.ListChildren(ctx, parentResult.Task.ID) + if err != nil { + t.Fatalf("ListChildren: %v", err) + } + if len(children) != 0 { + t.Fatalf("children after rollback = %#v, want none", children) + } +} + +func TestCreateChildTaskRollsBackWhenWorkspaceAttachmentIsUnavailable(t *testing.T) { + svc, repo := setupOfficeTest(t) + ctx := context.Background() + parentResult, err := svc.CreateTask(ctx, &CreateTaskRequest{ + WorkspaceID: "ws-1", Title: "Parent", ProjectID: "proj-1", + }) + if err != nil { + t.Fatalf("create parent: %v", err) + } + svc.SetWorkspacePolicyAttacher(nil) + + if _, err := svc.CreateChildTask(ctx, parentResult.Task, ChildTaskSpec{Title: "Child"}); err == nil { + t.Fatal("CreateChildTask succeeded without the required workspace attachment service") + } + children, err := repo.ListChildren(ctx, parentResult.Task.ID) + if err != nil { + t.Fatalf("ListChildren: %v", err) + } + if len(children) != 0 { + t.Fatalf("children after unavailable attachment rollback = %#v, want none", children) + } +} + func TestCreateChildTask_HappyPath_InheritsWorkflow(t *testing.T) { svc, repo := setupOfficeTest(t) ctx := context.Background() diff --git a/apps/backend/internal/task/service/service_detachment_test.go b/apps/backend/internal/task/service/service_detachment_test.go index e6aeecf3f9..45c5caf2fc 100644 --- a/apps/backend/internal/task/service/service_detachment_test.go +++ b/apps/backend/internal/task/service/service_detachment_test.go @@ -8,6 +8,7 @@ import ( "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -160,21 +161,6 @@ func TestDetachTaskPreservesDescendantsAndWorkspaceGroupMembership(t *testing.T) }); err != nil { t.Fatalf("CreateTask grandchild: %v", err) } - now := time.Now().UTC() - if _, err := repo.DB().ExecContext(ctx, ` - INSERT INTO task_workspace_groups - (id, workspace_id, owner_task_id, materialized_kind, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?) - `, "group-1", "workspace", "parent", "single_repo", now, now); err != nil { - t.Fatalf("insert workspace group: %v", err) - } - if _, err := repo.DB().ExecContext(ctx, ` - INSERT INTO task_workspace_group_members (workspace_group_id, task_id, role, created_at) - VALUES (?, ?, ?, ?) - `, "group-1", "child", "member", now); err != nil { - t.Fatalf("insert workspace group member: %v", err) - } - if _, err := svc.DetachTask(ctx, "child"); err != nil { t.Fatalf("DetachTask: %v", err) } @@ -198,6 +184,111 @@ func TestDetachTaskPreservesDescendantsAndWorkspaceGroupMembership(t *testing.T) } } +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.1 +func TestDetachTaskTransfersSharedWorkspaceStewardship(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + createDetachmentFixture(t, ctx, repo) + if _, err := repo.DB().ExecContext(ctx, ` + UPDATE task_workspace_groups SET materialized_environment_id = ? WHERE id = ? + `, "env-1", "group-1"); err != nil { + t.Fatalf("update workspace group environment: %v", err) + } + if err := repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ + ID: "env-1", TaskID: "parent", ExecutorType: string(models.ExecutorTypeWorktree), + WorkspacePath: t.TempDir(), Status: models.TaskEnvironmentStatusReady, + }); err != nil { + t.Fatalf("CreateTaskEnvironment: %v", err) + } + + if _, err := svc.DetachTask(ctx, "child"); err != nil { + t.Fatalf("DetachTask: %v", err) + } + + var groupOwner, environmentOwner string + var groupGeneration, environmentGeneration int64 + if err := repo.DB().QueryRowContext(ctx, + `SELECT owner_task_id, ownership_generation FROM task_workspace_groups WHERE id = ?`, "group-1", + ).Scan(&groupOwner, &groupGeneration); err != nil { + t.Fatalf("query workspace group owner: %v", err) + } + if err := repo.DB().QueryRowContext(ctx, + `SELECT task_id, ownership_generation FROM task_environments WHERE id = ?`, "env-1", + ).Scan(&environmentOwner, &environmentGeneration); err != nil { + t.Fatalf("query task environment owner: %v", err) + } + if groupOwner != "child" || environmentOwner != "child" { + t.Fatalf("owners after detach = group %q, environment %q; want child, child", groupOwner, environmentOwner) + } + var parentRole, childRole string + if err := repo.DB().QueryRowContext(ctx, ` + SELECT role FROM task_workspace_group_members WHERE workspace_group_id = ? AND task_id = ? + `, "group-1", "parent").Scan(&parentRole); err != nil { + t.Fatalf("query parent membership role: %v", err) + } + if err := repo.DB().QueryRowContext(ctx, ` + SELECT role FROM task_workspace_group_members WHERE workspace_group_id = ? AND task_id = ? + `, "group-1", "child").Scan(&childRole); err != nil { + t.Fatalf("query child membership role: %v", err) + } + if parentRole != "member" || childRole != "owner" { + t.Fatalf("membership roles after detach = parent %q, child %q; want member, owner", parentRole, childRole) + } + if groupGeneration != 2 || environmentGeneration != 2 { + t.Fatalf("generations after detach = group %d, environment %d; want 2, 2", groupGeneration, environmentGeneration) + } + + if _, err := svc.DetachTask(ctx, "child"); err != nil { + t.Fatalf("idempotent DetachTask: %v", err) + } + if err := repo.DB().QueryRowContext(ctx, + `SELECT ownership_generation FROM task_workspace_groups WHERE id = ?`, "group-1", + ).Scan(&groupGeneration); err != nil { + t.Fatalf("query workspace group generation after retry: %v", err) + } + if err := repo.DB().QueryRowContext(ctx, + `SELECT ownership_generation FROM task_environments WHERE id = ?`, "env-1", + ).Scan(&environmentGeneration); err != nil { + t.Fatalf("query task environment generation after retry: %v", err) + } + if groupGeneration != 2 || environmentGeneration != 2 { + t.Fatalf("generations after retry = group %d, environment %d; want 2, 2", groupGeneration, environmentGeneration) + } +} + +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.2 +func TestDetachTaskRollsBackHierarchyWhenWorkspaceTransferFails(t *testing.T) { + svc, _, repo := createTestService(t) + ctx := context.Background() + createDetachmentFixture(t, ctx, repo) + if _, err := repo.DB().ExecContext(ctx, ` + UPDATE task_workspace_groups SET materialized_environment_id = ? WHERE id = ? + `, "missing-environment", "group-1"); err != nil { + t.Fatalf("update workspace group environment: %v", err) + } + + if _, err := svc.DetachTask(ctx, "child"); err == nil { + t.Fatal("DetachTask succeeded with a missing canonical environment") + } + child, err := repo.GetTask(ctx, "child") + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if child.ParentID != "parent" { + t.Fatalf("child parent after rollback = %q, want parent", child.ParentID) + } + var owner string + var generation int64 + if err := repo.DB().QueryRowContext(ctx, ` + SELECT owner_task_id, ownership_generation FROM task_workspace_groups WHERE id = ? + `, "group-1").Scan(&owner, &generation); err != nil { + t.Fatalf("query workspace group after rollback: %v", err) + } + if owner != "parent" || generation != 1 { + t.Fatalf("workspace group after rollback = owner %q generation %d; want parent, 1", owner, generation) + } +} + func TestDetachTaskEventReflectsConcurrentReparent(t *testing.T) { svc, eventBus, repo := createTestService(t) ctx := context.Background() @@ -252,7 +343,7 @@ func TestDetachTaskPublishesOfficeTaskUpdated(t *testing.T) { t.Fatal("office.task.updated event was not published") } -func createDetachmentFixture(t *testing.T, ctx context.Context, repo taskEventTestRepository) { +func createDetachmentFixture(t *testing.T, ctx context.Context, repo *sqliterepo.Repository) { t.Helper() if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "workspace", Name: "Workspace"}); err != nil { t.Fatalf("CreateWorkspace: %v", err) @@ -280,6 +371,25 @@ func createDetachmentFixture(t *testing.T, ctx context.Context, repo taskEventTe }); err != nil { t.Fatalf("CreateTask child: %v", err) } + now := time.Now().UTC() + if _, err := repo.DB().ExecContext(ctx, ` + INSERT INTO task_workspace_groups + (id, workspace_id, owner_task_id, materialized_kind, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + `, "group-1", "workspace", "parent", "single_repo", now, now); err != nil { + t.Fatalf("insert workspace group: %v", err) + } + for _, member := range []struct { + taskID string + role string + }{{"parent", "owner"}, {"child", "member"}} { + if _, err := repo.DB().ExecContext(ctx, ` + INSERT INTO task_workspace_group_members (workspace_group_id, task_id, role, created_at) + VALUES (?, ?, ?, ?) + `, "group-1", member.taskID, member.role, now); err != nil { + t.Fatalf("insert workspace group member %s: %v", member.taskID, err) + } + } } func singleDetachmentEventData(t *testing.T, eventBus *MockEventBus) map[string]interface{} { diff --git a/apps/backend/internal/task/service/service_office_test.go b/apps/backend/internal/task/service/service_office_test.go index b8ba300d60..e5a9ba9e34 100644 --- a/apps/backend/internal/task/service/service_office_test.go +++ b/apps/backend/internal/task/service/service_office_test.go @@ -19,6 +19,12 @@ type dbStepResolver struct { repo *sqliterepo.Repository } +type noOpWorkspacePolicyAttacher struct{} + +func (noOpWorkspacePolicyAttacher) AttachWorkspacePolicy(context.Context, string, string, WorkspacePolicy) error { + return nil +} + func (r *dbStepResolver) ResolveStartStep(ctx context.Context, workflowID string) (string, error) { var stepID string err := r.repo.DB().QueryRowContext(ctx, @@ -83,6 +89,7 @@ func setupOfficeTest(t *testing.T) (*Service, *sqliterepo.Repository) { t.Fatalf("EnsureOfficeWorkflow: %v", err) } svc.SetStartStepResolver(&dbStepResolver{repo: repo}) + svc.SetWorkspacePolicyAttacher(noOpWorkspacePolicyAttacher{}) return svc, repo } diff --git a/apps/backend/internal/task/service/service_requests.go b/apps/backend/internal/task/service/service_requests.go index 947a4f069e..0794ad68f6 100644 --- a/apps/backend/internal/task/service/service_requests.go +++ b/apps/backend/internal/task/service/service_requests.go @@ -94,6 +94,9 @@ type CreateTaskRequest struct { ParentID string `json:"parent_id,omitempty"` Autopilot bool `json:"autopilot,omitempty"` WorkspacePath string `json:"workspace_path,omitempty"` // Optional host folder for repo-less tasks + // WorkspacePolicy carries the effective policy resolved by an API adapter. + // When omitted, CreateTask derives child defaults from the persisted parent. + WorkspacePolicy *WorkspacePolicy `json:"-"` // ExternalID is a caller-supplied identity used for create-idempotency // (docs/specs/tasks/requirements/external-id-idempotency.md). Accepted on REST diff --git a/apps/backend/internal/task/service/service_task_environments.go b/apps/backend/internal/task/service/service_task_environments.go index 9b885c59fe..e87fd27b1d 100644 --- a/apps/backend/internal/task/service/service_task_environments.go +++ b/apps/backend/internal/task/service/service_task_environments.go @@ -6,6 +6,7 @@ import ( "fmt" "strconv" + "github.com/google/uuid" "go.uber.org/zap" "github.com/kandev/kandev/internal/task/models" @@ -74,6 +75,15 @@ type SessionRunningChecker interface { IsAnySessionRunningForTask(ctx context.Context, taskID string) (bool, error) } +// taskEnvironmentResetClaimer reserves an environment reset behind the +// durable cleanup barrier. It is optional so lightweight service test doubles +// and integrations that do not persist cleanup jobs retain their legacy +// behavior. +type taskEnvironmentResetClaimer interface { + ClaimTaskEnvironmentReset(ctx context.Context, taskID, environmentID string, expectedGeneration int64, operationID string) (string, error) + ReleaseTaskEnvironmentReset(ctx context.Context, jobID string, state models.TaskResourceCleanupState, errStr string) error +} + // ErrSessionRunning is returned by ResetTaskEnvironment when at least one session // on the task is still actively running and the caller must stop it first. var ErrSessionRunning = errors.New("active session is running on this task; stop it before resetting the environment") @@ -295,6 +305,38 @@ func (s *Service) ResetTaskEnvironment(ctx context.Context, taskID string, opts return ErrSessionRunning } + var resetJobID string + var resetClaimer taskEnvironmentResetClaimer + if claimer, ok := s.taskEnvironments.(taskEnvironmentResetClaimer); ok { + resetClaimer = claimer + resetJobID, err = claimer.ClaimTaskEnvironmentReset( + ctx, + taskID, + env.ID, + env.OwnershipGeneration, + fmt.Sprintf("task-environment-reset:%s", uuid.NewString()), + ) + if err != nil { + return fmt.Errorf("claim environment reset: %w", err) + } + } + releaseReset := func(state models.TaskResourceCleanupState, releaseErr error) { + if resetClaimer == nil || resetJobID == "" { + return + } + lastError := "" + if releaseErr != nil { + lastError = releaseErr.Error() + } + if err := resetClaimer.ReleaseTaskEnvironmentReset(context.WithoutCancel(ctx), resetJobID, state, lastError); err != nil { + s.logger.Warn("failed to finalize task environment reset barrier", + zap.String("task_id", taskID), + zap.String("env_id", env.ID), + zap.String("cleanup_job_id", resetJobID), + zap.Error(err)) + } + } + s.logger.Info("resetting task environment", zap.String("task_id", taskID), zap.String("env_id", env.ID), @@ -306,20 +348,26 @@ func (s *Service) ResetTaskEnvironment(ctx context.Context, taskID string, opts if opts.PushBranch { if s.envDestroyer == nil { - return fmt.Errorf("environment destroyer not configured; cannot push branch") + err := fmt.Errorf("environment destroyer not configured; cannot push branch") + releaseReset(models.TaskResourceCleanupStateFailed, err) + return err } if err := s.envDestroyer.PushEnvironmentBranch(ctx, env); err != nil { + releaseReset(models.TaskResourceCleanupStateFailed, err) return fmt.Errorf("push branch before reset: %w", err) } } if err := s.teardownEnvironmentResources(ctx, env); err != nil { + releaseReset(models.TaskResourceCleanupStateFailed, err) return err } if err := s.taskEnvironments.DeleteTaskEnvironment(ctx, env.ID); err != nil { + releaseReset(models.TaskResourceCleanupStateFailed, err) return fmt.Errorf("delete task environment row: %w", err) } + releaseReset(models.TaskResourceCleanupStateSucceeded, nil) s.logger.Info("task environment reset complete", zap.String("task_id", taskID), zap.String("env_id", env.ID)) diff --git a/apps/backend/internal/task/service/service_task_environments_test.go b/apps/backend/internal/task/service/service_task_environments_test.go index f26a3b08da..cd004b45b3 100644 --- a/apps/backend/internal/task/service/service_task_environments_test.go +++ b/apps/backend/internal/task/service/service_task_environments_test.go @@ -7,6 +7,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/worktree" ) type stubEnvRepo struct { @@ -96,6 +97,37 @@ type stubRunningChecker struct { err error } +// @covers AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.4 +func TestCleanupTaskEnvironmentSkipsStaleOwnershipGeneration(t *testing.T) { + repo := &stubEnvRepo{env: &models.TaskEnvironment{ + ID: "env-transferred", TaskID: "detached-child", OwnershipGeneration: 2, + }} + svc := newResetTestService(t, repo) + destroyer := &stubDestroyer{} + svc.SetEnvironmentDestroyer(destroyer) + worktreeCleaner := &recordingWorktreeCleanup{} + svc.SetWorktreeCleanup(worktreeCleaner) + + errs := svc.cleanupDestructiveTaskResources( + context.Background(), "former-parent", nil, + []*worktree.Worktree{{ID: "replacement-worktree"}}, + taskEnvironmentCleanup{ + env: &models.TaskEnvironment{ + ID: "env-transferred", TaskID: "former-parent", OwnershipGeneration: 1, + ContainerID: "replacement-container", + }, + deleteRow: true, + }, nil) + + if len(errs) != 0 { + t.Fatalf("cleanup errors = %v, want stale snapshot no-op", errs) + } + if len(destroyer.containerCalls) != 0 || len(worktreeCleaner.cleaned) != 0 || repo.deleted { + t.Fatalf("stale cleanup mutated replacement: container calls %v, worktree calls %v, row deleted %v", + destroyer.containerCalls, worktreeCleaner.cleaned, repo.deleted) + } +} + func (s *stubRunningChecker) IsAnySessionRunningForTask(context.Context, string) (bool, error) { return s.running, s.err } diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index ad4dbdd56f..0e26b3cb29 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -101,7 +101,12 @@ type taskEnvironmentSessionBorrowerFinder interface { } type taskEnvironmentOwnerTransferer interface { - TransferTaskEnvironmentToTask(ctx context.Context, envID, taskID string) error + TransferTaskEnvironmentOwnership( + ctx context.Context, + envID, expectedTaskID string, + expectedGeneration int64, + taskID string, + ) error } type workflowStepCapacityTaskCreator interface { @@ -249,6 +254,9 @@ func (s *Service) prepareTaskForCreation(ctx context.Context, req *CreateTaskReq if err := s.inheritParentProject(ctx, req); err != nil { return nil, err } + if err := s.prepareWorkspacePolicyForCreation(ctx, req); err != nil { + return nil, err + } if err := prepareAutoTitle(req); err != nil { return nil, err @@ -390,6 +398,15 @@ func (s *Service) finalizeCreatedTask(ctx context.Context, prepared *preparedTas if err := s.persistTaskRepositoryRows(ctx, task.ID, prepared.repositories); err != nil { return CreateTaskResult{}, s.rollbackPartialTask(ctx, task.ID, err) } + if req.WorkspacePolicy != nil && req.WorkspacePolicy.NeedsAttachment() { + if s.workspacePolicyAttacher == nil { + return CreateTaskResult{}, s.rollbackPartialTask(ctx, task.ID, + errors.New("workspace policy attachment is unavailable")) + } + if err := s.workspacePolicyAttacher.AttachWorkspacePolicy(ctx, task.ID, req.ParentID, *req.WorkspacePolicy); err != nil { + return CreateTaskResult{}, s.rollbackPartialTask(ctx, task.ID, fmt.Errorf("attach workspace policy: %w", err)) + } + } // Load repositories into task for response repos, err := s.taskRepos.ListTaskRepositories(ctx, task.ID) @@ -412,6 +429,76 @@ func (s *Service) finalizeCreatedTask(ctx context.Context, prepared *preparedTas return CreateTaskResult{Task: task, Outcome: CreateTaskOutcomeCreated}, nil } +func (s *Service) prepareWorkspacePolicyForCreation(ctx context.Context, req *CreateTaskRequest) error { + policy := WorkspacePolicy{} + if req.WorkspacePolicy != nil { + policy = *req.WorkspacePolicy + } else if workspace, ok := req.Metadata["workspace"].(map[string]interface{}); ok { + policy.Mode, _ = workspace["mode"].(string) + policy.GroupID, _ = workspace["group_id"].(string) + policy.DefaultChildWorkspace, _ = workspace["default_child_workspace"].(string) + policy.DefaultChildOrdering, _ = workspace["default_child_ordering"].(string) + } + if err := s.applyParentWorkspacePolicyDefaults(ctx, req, &policy); err != nil { + return err + } + if err := validateWorkspacePolicy(req.ParentID, &policy); err != nil { + return err + } + if len(policy.MetadataBlock()) == 0 { + return nil + } + req.Metadata = policy.MergeMetadataBlock(req.Metadata) + req.WorkspacePolicy = &policy + return nil +} + +func (s *Service) applyParentWorkspacePolicyDefaults( + ctx context.Context, + req *CreateTaskRequest, + policy *WorkspacePolicy, +) error { + if req.ParentID == "" { + return nil + } + parent, err := s.tasks.GetTask(ctx, req.ParentID) + if err != nil { + return fmt.Errorf("load parent workspace policy: %w", err) + } + if parent == nil { + return fmt.Errorf("parent task %s not found", req.ParentID) + } + parentWorkspace, _ := parent.Metadata["workspace"].(map[string]interface{}) + if policy.ParentOrdering == "" { + policy.ParentOrdering, _ = parentWorkspace["default_child_ordering"].(string) + } + if policy.Mode != "" { + return nil + } + policy.Mode, _ = parentWorkspace["default_child_workspace"].(string) + return nil +} + +func validateWorkspacePolicy(parentID string, policy *WorkspacePolicy) error { + switch policy.Mode { + case "": + return nil + case workspaceModeInheritParent: + if parentID == "" { + return fmt.Errorf("workspace_mode=%s requires parent_id", workspaceModeInheritParent) + } + case workspaceModeNewWorkspace: + policy.GroupID = "" + case workspaceModeSharedGroup: + if policy.GroupID == "" { + return errors.New("workspace_group_id is required when workspace_mode=shared_group") + } + default: + return fmt.Errorf("invalid workspace_mode: %s", policy.Mode) + } + return nil +} + // findTaskByExternalIDIfPresent is CreateTask's step-3 lookup: when the // request carries an external_id already held by a task, the caller MUST // return that task (found=true) instead of continuing the create. found is @@ -3345,20 +3432,36 @@ func (s *Service) cleanupDestructiveTaskResources( if cause := context.Cause(ctx); cause != nil { return []error{cause} } - skipOwnedEnvironment, err := s.hasActiveOtherTaskSessionsForEnvironment(ctx, taskID, envCleanup.env) + currentOwnership, err := s.taskEnvironmentCleanupOwnershipIsCurrent(ctx, envCleanup.env) if err != nil { - s.logger.Warn("skipping task environment cleanup after shared-environment ownership check failed", + s.logger.Warn("skipping task environment cleanup after ownership-generation check failed", zap.String("task_id", taskID), zap.String("env_id", taskEnvironmentID(envCleanup.env)), zap.Error(err)) errs = append(errs, fmt.Errorf("check task environment ownership %s: %w", taskEnvironmentID(envCleanup.env), err)) - skipOwnedEnvironment = true + } + skipOwnedEnvironment := !currentOwnership + if currentOwnership { + var shared bool + shared, err = s.hasActiveOtherTaskSessionsForEnvironment(ctx, taskID, envCleanup.env) + skipOwnedEnvironment = shared + if err != nil { + s.logger.Warn("skipping task environment cleanup after shared-environment ownership check failed", + zap.String("task_id", taskID), + zap.String("env_id", taskEnvironmentID(envCleanup.env)), + zap.Error(err)) + errs = append(errs, fmt.Errorf("check task environment ownership %s: %w", taskEnvironmentID(envCleanup.env), err)) + skipOwnedEnvironment = true + } } if skipOwnedEnvironment { s.logger.Info("skipping task environment cleanup while another task still uses it", zap.String("task_id", taskID), zap.String("env_id", taskEnvironmentID(envCleanup.env))) } + if !currentOwnership { + return errs + } if cause := context.Cause(ctx); cause != nil { return append(errs, cause) } @@ -3488,7 +3591,9 @@ func (s *Service) preserveTaskEnvironmentForActiveBorrower(ctx context.Context, if !ok { return false, fmt.Errorf("task environment repository cannot transfer borrowed environment %s", env.ID) } - if err := ownerTransfer.TransferTaskEnvironmentToTask(ctx, env.ID, borrowerTaskID); err != nil { + if err := ownerTransfer.TransferTaskEnvironmentOwnership( + ctx, env.ID, env.TaskID, env.OwnershipGeneration, borrowerTaskID, + ); err != nil { return false, fmt.Errorf("transfer task environment %s to %s: %w", env.ID, borrowerTaskID, err) } env.TaskID = borrowerTaskID @@ -3608,6 +3713,13 @@ func (s *Service) cleanupTaskEnvironment( if cause := context.Cause(ctx); cause != nil { return []error{cause} } + current, err := s.taskEnvironmentCleanupOwnershipIsCurrent(ctx, cleanup.env) + if err != nil { + return []error{fmt.Errorf("verify task environment ownership %s: %w", cleanup.env.ID, err)} + } + if !current { + return nil + } if err := s.teardownEnvironmentResources(ctx, cleanup.env); err != nil { s.logger.Warn("failed to teardown task environment during task cleanup", zap.String("task_id", taskID), @@ -3631,6 +3743,63 @@ func (s *Service) cleanupTaskEnvironment( return nil } +func (s *Service) taskEnvironmentCleanupOwnershipIsCurrent( + ctx context.Context, + snapshot *models.TaskEnvironment, +) (bool, error) { + if snapshot == nil || snapshot.ID == "" { + return true, nil + } + if s.taskEnvironments == nil { + return false, errors.New("task environment repository is unavailable") + } + current, err := s.taskEnvironments.GetTaskEnvironment(ctx, snapshot.ID) + if errors.Is(err, taskrepo.ErrTaskEnvironmentNotFound) { + return s.taskEnvironmentSnapshotOwnerDeleted(ctx, snapshot) + } + if err != nil { + return false, err + } + if current == nil { + return s.taskEnvironmentSnapshotOwnerDeleted(ctx, snapshot) + } + if current.TaskID == snapshot.TaskID && + (snapshot.OwnershipGeneration == 0 || current.OwnershipGeneration == snapshot.OwnershipGeneration) { + return true, nil + } + s.logger.Info("skipping stale task environment cleanup", + zap.String("env_id", snapshot.ID), + zap.String("expected_owner_task_id", snapshot.TaskID), + zap.Int64("expected_ownership_generation", snapshot.OwnershipGeneration), + zap.String("current_owner_task_id", current.TaskID), + zap.Int64("current_ownership_generation", current.OwnershipGeneration)) + return false, nil +} + +// taskEnvironmentSnapshotOwnerDeleted proves that a missing environment row +// is the result of the owning task being deleted, rather than treating a +// missing authority row as permission to use an arbitrary stale snapshot. +func (s *Service) taskEnvironmentSnapshotOwnerDeleted( + ctx context.Context, + snapshot *models.TaskEnvironment, +) (bool, error) { + if snapshot == nil || snapshot.TaskID == "" || s.tasks == nil { + if snapshot != nil && snapshot.TaskID == "" && + snapshot.ContainerID == "" && snapshot.SandboxID == "" { + return true, nil + } + return false, nil + } + task, err := s.tasks.GetTask(ctx, snapshot.TaskID) + if errors.Is(err, taskrepo.ErrTaskNotFound) || task == nil { + return true, nil + } + if err != nil { + return false, err + } + return false, nil +} + // ListTasks returns all tasks for a workflow func (s *Service) ListTasks(ctx context.Context, workflowID string) ([]*models.Task, error) { if err := s.authorizeWorkflowID(ctx, workflowID); err != nil { diff --git a/docs/decisions/2026-09-04-generation-fenced-task-environment-ownership.md b/docs/decisions/2026-09-04-generation-fenced-task-environment-ownership.md new file mode 100644 index 0000000000..6204d9fdfb --- /dev/null +++ b/docs/decisions/2026-09-04-generation-fenced-task-environment-ownership.md @@ -0,0 +1,71 @@ +# ADR-2026-09-04-generation-fenced-task-environment-ownership: Fence Task Environment Ownership by Generation + +**Status:** accepted +**Date:** 2026-09-04 +**Area:** backend + +## Context + +ADR-2026-08-08 makes `task_environments.task_id` the canonical owner of a task +workspace and permits shared-environment ownership transfer. Subtask detachment +currently changes only `tasks.parent_id` and workspace-mode metadata. The shared +workspace group and materialized environment remain owned by the former parent. + +Archive and delete cleanup persist resource snapshots and can retry after a +restart. An environment ID or path does not prove that the task named by an old +snapshot still owns that resource. A delayed cleanup can therefore act after +ownership has moved, while handler-specific creation attachment can leave some +child tasks outside the workspace group entirely. + +## Decision + +Persist a monotonically increasing ownership generation on every task +environment and shared workspace group. Treat the tuple of resource ID, owner +task ID, and ownership generation as the authority for ownership transfer and +destructive cleanup. + +Detachment of a task using a shared workspace changes hierarchy, workspace-group +stewardship, membership roles, environment owner, and both ownership generations +in one database transaction. The transaction uses the same task cleanup barrier +as archive and delete so detachment either precedes cleanup inventory or is +rejected after cleanup has been admitted. + +Every owner transfer is a compare-and-swap against the expected owner and +generation and increments the generation. Cleanup claims the expected +generation for the duration of external teardown; transfers reject an active +claim. Stale cleanup, retry, rollback, and status updates fail closed when the +owner or generation differs. + +Workspace-policy attachment is required synchronous task-creation work owned by +the task service. API handlers and internal callers do not implement separate +membership attachment sequences. + +## Consequences + +- Parent lifecycle operations cannot destroy a detached child's current + environment after stewardship transfers. +- Durable cleanup snapshots remain safe across retries and backend restarts. +- SQLite and PostgreSQL require replayable schema changes and dialect-specific + concurrency tests. +- Ownership-transfer callers must carry expected owner and generation instead + of issuing unguarded owner updates. +- Group cleanup needs a generation-scoped claim around filesystem or provider + operations. +- Task creation does not report a child as complete until required workspace + membership exists. + +## Alternatives Considered + +1. **Preserve group membership without changing ownership.** Rejected because + task cleanup authority still belongs to the former parent. +2. **Copy the physical workspace during detachment.** Rejected because it cannot + atomically move active sessions, is expensive for multi-repository tasks, and + changes the established shared-workspace behavior. +3. **Check the current owner immediately before cleanup without a generation.** + Rejected because ownership can change after the check and before the external + destructive operation. +4. **Serialize with in-process mutexes.** Rejected because process restart and + PostgreSQL multi-connection execution are durable concurrency boundaries. +5. **Keep attachment in each API handler.** Rejected because WebSocket, plugin, + workflow, and future internal task creation can bypass handler-owned logic. + diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 7f5221b751..20bf5153b4 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -237,4 +237,5 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 2026-09-01-passthrough-initial-prompt-turn-boundary | [Keep Passthrough Initial Prompt State in Lifecycle](2026-09-01-passthrough-initial-prompt-turn-boundary.md) | accepted | backend, workflow | 2026-09-01 | | 2026-09-02-automation-self-archive | [Permit Automation Self-Archive as Terminal Completion](2026-09-02-automation-self-archive.md) | accepted | backend, agentctl, protocol, security, workflow | 2026-09-02 | | 2026-09-03-separate-system-data-storage-pages | [Separate System Data and Storage Pages](2026-09-03-separate-system-data-storage-pages.md) | accepted | frontend | 2026-09-03 | +| 2026-09-04-generation-fenced-task-environment-ownership | [Fence Task Environment Ownership by Generation](2026-09-04-generation-fenced-task-environment-ownership.md) | accepted | backend | 2026-09-04 | | 2026-09-04-use-repository-token-for-runtime-pin-prs | [Use the built-in Actions token for runtime pin PRs](2026-09-04-use-repository-token-for-runtime-pin-prs.md) | accepted | workflow, security | 2026-09-04 | diff --git a/docs/plans/detached-workspace-continuity/plan.md b/docs/plans/detached-workspace-continuity/plan.md new file mode 100644 index 0000000000..68b110ec18 --- /dev/null +++ b/docs/plans/detached-workspace-continuity/plan.md @@ -0,0 +1,142 @@ +--- +created: 2026-09-04 +status: completed +requirements: + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-001 + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-002 +system_design: + - ../../specs/tasks/system-design/detached-workspace-continuity.md +legacy_specs: + - ../../specs/tasks/requirements/subtask-detachment.md +--- + +# Implementation Plan: Detached Workspace Continuity + +## Overview + +Make detachment transfer shared-workspace stewardship atomically, route every +child creation surface through the same workspace attachment step, and prevent +stale cleanup from acting after an ownership generation changes. The work is one +sequential backend package because the schema, repository transaction, service +flow, and regression tests share one ownership contract. + +## Scope + +### In scope + +- Persist ownership generations for workspace groups and task environments. +- Atomically detach a task and transfer group and environment stewardship. +- Fence task-resource and workspace-group cleanup by owner and generation. +- Move child workspace attachment into the canonical task creation sequence. +- Cover SQLite, PostgreSQL, concurrency, restart recovery, and all creation + entry points named by the requirements. + +### Out of scope + +- Copying a shared workspace during detachment. +- Frontend interaction changes or new user-facing copy. +- Changing workflow placement, descendant relationships, blockers, or active + session lifecycle during detachment. +- Broad task cleanup refactoring beyond ownership and generation checks. + +## Technical approach + +### Persistence and models + +- Add `OwnershipGeneration` to Office `WorkspaceGroup` and task + `TaskEnvironment` models. +- Add replayable `ownership_generation` migrations and fresh-schema columns in + the task and Office SQLite/PostgreSQL repositories. +- Extend environment and group queries, inserts, updates, and cleanup snapshots + to preserve the generation. + +### Atomic ownership operations + +- Replace the repository's narrow `DetachTask` update with a transaction that + locks child, former parent, active group, and materialized environment; checks + cleanup barriers; updates hierarchy, roles, owners, and generations; and + commits once. +- Replace unguarded environment owner transfers with expected-owner and + expected-generation compare-and-swap operations used by direct delete, + cascade archive/delete, rollback, and detachment. +- Add generation-aware workspace-group cleanup claim and completion methods so + stale cleanup cannot change status or remove resources. + +### Canonical creation + +- Add workspace-policy inputs and an injected attachment coordinator to the task + service create sequence. +- Resolve and attach policy before `task.created` publication and before Created + returns; reuse partial-task rollback on failure. +- Remove handler-owned REST and MCP attachment calls and cover WebSocket, + plugin, workflow/internal child, REST, and MCP creation through the service + boundary. + +### Cleanup and recovery + +- Compare live environment owner and generation with the cleanup snapshot before + task environment and worktree teardown. +- Treat superseded snapshots and retries as safe no-ops while preserving error + reporting for genuinely failed current-generation cleanup. +- Prove that restart processing cannot revive an old generation's authority. + +## Tests + +- `apps/backend/internal/task/repository/sqlite/detached_workspace_continuity_test.go`: + SQLite transaction, idempotency, rollback, cleanup-barrier, concurrent detach, + and stale-generation cases. +- `apps/backend/internal/task/repository/sqlite/detached_workspace_continuity_postgres_test.go`: + PostgreSQL row-lock serialization and fresh/replayed migration coverage. +- `apps/backend/internal/office/repository/sqlite/workspace_group_generation_test.go`: + group generation claims, membership role uniqueness, stale status writes, and + SQLite/PostgreSQL schema replay. +- `apps/backend/internal/task/service/service_detachment_test.go` and a focused + cleanup test file: detached child continuity after parent archive/delete, + stale durable-job recovery, and unchanged physical workspace binding. +- Handler, MCP, plugin adapter, and `CreateChildTask` tests: every creation route + reaches the service-owned workspace attachment coordinator and rolls back on + failure. + +## E2E tests + +Skipped. The existing detach UI and API response do not change. Repository and +service integration tests exercise the workspace lifecycle guarantee below the +UI. + +## Work orders + +- [x] [Task 01: Transfer workspace ownership](task-01-transfer-ownership.md) *(completed)* + +## Verification results + +Completed on 2026-09-04. + +- Atomic detach, stewardship transfer, role updates, and ownership generations + are covered by service and repository regressions. +- Current-owner cleanup barriers, guarded owner-generation transfers, stale + environment cleanup, and generation-fenced workspace cleanup are covered. +- The task service now owns workspace attachment before publishing or returning + a newly created task; attachment failure rolls the task row back. +- Focused repository tests passed: 27 tests across task and Office persistence. +- Focused service tests passed: 19 tests. +- Creation entry-point tests passed: 15 tests across HTTP, MCP, plugins, and + backend composition. +- Final combined ownership and creation regressions passed: 76 tests across + seven packages. +- Race-enabled ownership tests passed: 37 tests across three packages. +- Backend lint passed with zero issues. +- Specification linter tests passed: 30 tests; all specification files passed. +- PostgreSQL integration tests were not run because no isolated + `KANDEV_TEST_POSTGRES_DSN` was configured. +- The full backend suite reported 9,907 passes, 16 skips, and seven unrelated + config-discovery failures caused by the VM's existing + `/root/.kandev/config.yaml` overriding those tests' temporary home fixtures. + +## Risks + +- Task and Office repositories initialize tables in a defined order; both fresh + and replayed schemas must converge before cross-table detachment queries run. +- Cleanup invokes external filesystem or provider operations, so its database + claim must prevent ownership transfer for the complete destructive window. +- External-ID deduplication must not attach a retry's workspace policy to an + existing task. diff --git a/docs/plans/detached-workspace-continuity/task-01-transfer-ownership.md b/docs/plans/detached-workspace-continuity/task-01-transfer-ownership.md new file mode 100644 index 0000000000..704e993e2d --- /dev/null +++ b/docs/plans/detached-workspace-continuity/task-01-transfer-ownership.md @@ -0,0 +1,148 @@ +--- +id: "01-transfer-ownership" +title: "Transfer workspace ownership" +status: completed +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-001 + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-002 +acceptance_criteria: + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.1 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.2 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.3 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.4 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.5 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.1 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.2 + - AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.3 +system_design: + - ../../specs/tasks/system-design/detached-workspace-continuity.md +--- + +# Task 01: Transfer Workspace Ownership + +## Summary + +Implement generation-fenced workspace stewardship from schema through task +creation, detachment, cleanup, and restart recovery. The result keeps the +detached child's canonical environment alive after former-parent lifecycle +operations and makes every child-creation route establish the same membership +before launch. + +## In scope + +- Add replayable workspace-group and task-environment ownership generations. +- Implement atomic detachment and guarded owner-transfer repository operations. +- Fence task-resource and workspace-group cleanup by current owner, generation, + membership, and cleanup barrier. +- Centralize child workspace-policy resolution and attachment in `CreateTask`. +- Add focused SQLite, PostgreSQL, service, concurrency, restart, and entry-point + regression tests. + +## Out of scope + +- Workspace copying, session restart, or frontend behavior changes. +- Cleanup changes unrelated to shared ownership. +- New task hierarchy or blocker behavior. + +## Acceptance + +- Detachment either commits hierarchy, group steward, membership roles, + environment owner, and incremented generations together or commits none of + them. +- Parent archive/delete and delayed cleanup cannot stop, delete, reuse, or + change a detached child's current-generation workspace. +- Every supported child creation route attaches the canonical workspace policy + before publishing or returning a launchable task, with rollback on failure. + +## Verification + +```bash +cd apps/backend && go test ./internal/task/repository/sqlite ./internal/office/repository/sqlite -run 'Test.*(DetachedWorkspaceContinuity|OwnershipGeneration|WorkspaceGroupGeneration|SchemaReplay)' -count=1 +cd apps/backend && go test ./internal/task/service -run 'Test.*(Detach|DetachedWorkspace|Stale.*Cleanup|CreateChildTask.*Workspace)' -count=1 +cd apps/backend && go test ./internal/task/handlers ./internal/mcp/handlers ./internal/plugins ./internal/backendapp -run 'Test.*CreateTask.*Workspace' -count=1 +KANDEV_TEST_POSTGRES_DSN='' go test ./internal/task/repository/sqlite ./internal/office/repository/sqlite -run 'Test.*(DetachedWorkspaceContinuity|OwnershipGeneration|WorkspaceGroupGeneration|SchemaReplay).*Postgres' -count=1 +``` + +## Files likely touched + +- `apps/backend/internal/task/models/models.go` +- `apps/backend/internal/office/models/workspace_groups.go` +- `apps/backend/internal/task/repository/interface.go` +- `apps/backend/internal/task/repository/sqlite/base_schema.go` +- `apps/backend/internal/task/repository/sqlite/base_migrations.go` +- `apps/backend/internal/task/repository/sqlite/task.go` +- `apps/backend/internal/task/repository/sqlite/task_environment.go` +- `apps/backend/internal/task/repository/sqlite/workspace_group.go` +- `apps/backend/internal/office/repository/sqlite/base_migrations.go` +- `apps/backend/internal/office/repository/sqlite/workspace_groups.go` +- `apps/backend/internal/task/service/service.go` +- `apps/backend/internal/task/service/service_tasks.go` +- `apps/backend/internal/task/service/service_detachment.go` +- `apps/backend/internal/task/service/resource_cleanup_jobs.go` +- `apps/backend/internal/task/service/handoff_service.go` +- `apps/backend/internal/task/service/handoff_cascade.go` +- `apps/backend/internal/task/service/handoff_cleaner.go` +- `apps/backend/internal/task/handlers/task_http_handlers.go` +- `apps/backend/internal/task/handlers/task_ws_handlers.go` +- `apps/backend/internal/mcp/handlers/handlers.go` +- `apps/backend/internal/backendapp/helpers.go` +- Focused adjacent `*_test.go` files. + +## Dependencies + +None. + +## Risks + +- Cross-table lock order must remain identical in detach, lifecycle cleanup, + and ownership-transfer paths to avoid PostgreSQL deadlocks. +- A cleanup claim that is released too early reopens the external-operation + race; failure paths must retain or explicitly release the same generation. +- Existing tests and fakes implement narrow repository interfaces and will need + compatible generation-aware methods. + +## Parallelism + +`sequential` + +## Inputs + +- `docs/specs/tasks/requirements/detached-workspace-continuity.md` +- `docs/specs/tasks/system-design/detached-workspace-continuity.md` +- `docs/decisions/2026-09-04-generation-fenced-task-environment-ownership.md` +- Existing detachment, handoff, task-resource cleanup, and task-create tests. + +## Results + +Implemented generation-fenced workspace ownership end to end: + +- Added replayable ownership generations to workspace groups and task + environments, including fresh schema and worktree cutover paths. +- Made detachment transactionally update hierarchy, workspace mode, group + steward, membership roles, environment owner, and both generations. +- Added cleanup-barrier-aware, expected-owner and expected-generation + environment transfers; removed the unguarded transfer API. +- Added generation claims for workspace-group cleanup and stale-snapshot guards + for the complete environment and worktree cleanup pipeline. +- Moved workspace-policy attachment into `Service.CreateTask`, wired all + production entry points through it, and retained rollback on attachment + failure. +- Added focused tests for atomic rollback, idempotency, ownership roles, + generation advancement, stale retry rejection, cleanup fencing, and internal + child creation. + +Verification results: + +- Repository focus: 27 passed. +- Service focus: 19 passed. +- HTTP/MCP/plugin/backend entry-point focus: 15 passed. +- Final combined ownership and creation focus: 76 passed across seven packages. +- Race-enabled ownership focus: 37 passed across three packages. +- Backend lint: zero issues. +- Spec linter: 30 tests passed and all spec files passed. +- PostgreSQL-only run: skipped because no isolated DSN was available. +- Full backend run: 9,907 passed, 16 skipped, seven environment-sensitive + config tests failed because this VM has a real `/root/.kandev/config.yaml`. diff --git a/docs/specs/tasks/README.md b/docs/specs/tasks/README.md index 7b43f38a0d..5d45b25036 100644 --- a/docs/specs/tasks/README.md +++ b/docs/specs/tasks/README.md @@ -46,6 +46,7 @@ signals, and task-scoped scheduling contracts. - [Active clarification lifecycle scenarios](requirements/clarification-active-lifecycle-scenarios.md) - [Active clarification lifecycle](requirements/clarification-active-lifecycle.md) - [Task Documents](requirements/documents.md) +- [Detached Workspace Continuity](requirements/detached-workspace-continuity.md) - [Task Execution Stages](requirements/execution-stages.md) - [External task ID idempotency boundaries](requirements/external-id-idempotency-boundaries.md) - [External task ID idempotency scenarios](requirements/external-id-idempotency-scenarios.md) @@ -122,6 +123,7 @@ signals, and task-scoped scheduling contracts. - [Environment-Owned Git Status](system-design/environment-owned-git-status.md) - [Attach Workspace Sources](system-design/attach-workspace-sources.md) - [Command-panel archived task results](system-design/command-panel-archived-task-results.md) +- [Detached Workspace Continuity](system-design/detached-workspace-continuity.md) - [Quick Chat Agent Titles](system-design/quick-chat-agent-titles.md) - [Quick Chat Session Resumption](system-design/quick-chat-session-resumption.md) - [Active clarification lifecycle](system-design/clarification-active-lifecycle.md) diff --git a/docs/specs/tasks/requirements/detached-workspace-continuity.md b/docs/specs/tasks/requirements/detached-workspace-continuity.md new file mode 100644 index 0000000000..9a8da6d845 --- /dev/null +++ b/docs/specs/tasks/requirements/detached-workspace-continuity.md @@ -0,0 +1,82 @@ +--- +status: active +system: tasks +created: 2026-09-04 +owners: + - kandev +--- + +# Detached Workspace Continuity Requirements + +## Overview + +The task system owns the lifetime of task environments and task-owned +worktrees. Detaching a subtask must therefore preserve the workspace that its +sessions use and must separate that workspace's cleanup authority from the +former parent's lifecycle. + +## Terminology + +- **Workspace steward:** The current task whose identity owns a shared + workspace group and its canonical task environment. Stewardship can move + between active group members without changing the physical workspace. +- **Ownership generation:** A monotonically increasing value that identifies + one assignment of a workspace group or task environment to a steward. +- **Stale cleanup:** Cleanup prepared for an owner or ownership generation that + is no longer current. + +## Requirements + +### REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-001: Detached task workspace continuity + +**Intent:** Keep a detached task's running and resumable work intact when the +former parent is archived, deleted, retried, or recovered after restart. + +#### Acceptance criteria + +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.1:** When a subtask that uses its + parent's shared workspace is detached, the system shall make an active member + of that shared workspace independent of the former parent's cleanup authority + before reporting detachment success. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.2:** When detachment changes + workspace stewardship, the hierarchy change and workspace-group and + environment ownership changes shall either all become durable or none shall + become durable. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.3:** When the former parent is + archived or deleted after detachment, the detached task shall continue using + the same canonical workspace without its active or resumable environment + being stopped, removed, or made reusable by another task. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.4:** When cleanup or an ownership + retry presents an owner or ownership generation that is no longer current, + the system shall reject the destructive operation without changing the + replacement workspace, environment, or task. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-001.5:** When the backend restarts + after detachment, the detached task shall recover the same workspace binding, + current steward, and ownership generation. + +### REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-002: Canonical subtask workspace attachment + +**Intent:** Ensure every task-creation surface establishes the same durable +workspace relationship before a child can launch or be returned as completely +created. + +#### Acceptance criteria + +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.1:** When any supported REST, + WebSocket, MCP, plugin, workflow, or internal task-creation route creates a + child task, the system shall apply the same parent workspace-policy resolution + and membership attachment behavior. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.2:** When workspace attachment + fails, the task creation shall fail without leaving a launchable child that + lacks its required workspace membership. +- **AC-TASKS-DETACHED-WORKSPACE-CONTINUITY-002.3:** When child creation, + detachment, parent cleanup, and workspace stewardship changes overlap, the + system shall serialize them or reject one operation without a partial + hierarchy or ownership result. + +## Out of scope + +- Copying or provisioning a second physical workspace during detachment. +- Stopping or restarting the detached task's active sessions. +- Changing descendants, blockers, workflow placement, or repository selection. +- Allowing multiple physical owners for one canonical task environment. diff --git a/docs/specs/tasks/requirements/subtask-detachment.md b/docs/specs/tasks/requirements/subtask-detachment.md index e0676ad4db..289c888bf4 100644 --- a/docs/specs/tasks/requirements/subtask-detachment.md +++ b/docs/specs/tasks/requirements/subtask-detachment.md @@ -31,7 +31,7 @@ Users can create and reorganize task trees, but they cannot promote an existing - A user can detach a subtask from its parent from the sidebar task context menu and the task card's three-dot and context menus. - The action is shown only for tasks with a parent and requires confirmation. -- Confirmation explains that detachment changes task hierarchy only and that a workspace shared with the parent remains shared. +- Confirmation explains that the workspace remains shared even though cleanup stewardship is separated from the former parent. - Detachment clears the task's parent relationship without changing its workflow, workflow step, state, blockers, descendants, repositories, sessions, or agent execution. - A detached task becomes a top-level root of its existing subtree. Its descendants remain attached to it. - A task using `inherit_parent` workspace mode changes to `shared_group` mode when detached. Its workspace-group membership remains active, so current and future sessions continue using the same materialized workspace. @@ -46,8 +46,9 @@ Detachment updates existing persisted fields; it adds no table or column. - `tasks.parent_id` becomes the empty string. - `tasks.metadata.workspace.mode` changes from `inherit_parent` to `shared_group` when applicable. -- `task_workspace_group_members` is unchanged. Active membership remains the durable source of shared workspace access. -- Existing blocker rows, task-session rows, task-environment rows, and descendant `parent_id` values are unchanged. +- The child's active `task_workspace_group_members` row remains the durable source of shared workspace access. The owner role can move to the child. +- Workspace-group and task-environment owners and ownership generations change atomically when the former parent owns the shared materialization. +- Existing blocker rows, task-session rows, environment identity, and descendant `parent_id` values are unchanged. The hierarchy update and workspace-mode update are persisted in the same task-row update. @@ -75,11 +76,11 @@ If workspace mode is `inherit_parent`, the same operation also transitions it to - If the task no longer exists, the UI keeps its current state and shows the request error. - If persistence fails, neither the UI nor API reports success. The task remains attached according to durable state. - Repeated submissions are safe because detaching an already-root task is a no-op. -- Detachment does not create, copy, move, clean, or delete workspace files. +- Detachment does not create, copy, move, clean, or delete workspace files. It can transfer the logical stewardship of those files. ## Persistence guarantees -The cleared parent relationship and normalized workspace mode survive backend restarts. Existing workspace-group membership and materialized workspace ownership remain unchanged and follow the existing cleanup lifecycle for the group. +The cleared parent relationship, normalized workspace mode, current workspace steward, and ownership generation survive backend restarts. See [Detached Workspace Continuity](detached-workspace-continuity.md) for the cleanup and concurrency contract. ## Scenarios @@ -105,4 +106,4 @@ The cleared parent relationship and normalized workspace mode survive backend re ## Implementation plan -See [the implementation plan](../../../plans/subtask-detachment/plan.md). \ No newline at end of file +See [the implementation plan](../../../plans/subtask-detachment/plan.md). diff --git a/docs/specs/tasks/system-design/detached-workspace-continuity.md b/docs/specs/tasks/system-design/detached-workspace-continuity.md new file mode 100644 index 0000000000..0c7f168af3 --- /dev/null +++ b/docs/specs/tasks/system-design/detached-workspace-continuity.md @@ -0,0 +1,188 @@ +--- +status: current +system: tasks +requirements: + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-001 + - REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-002 +--- + +# Detached Workspace Continuity System Design + +## Purpose and boundaries + +Task lifecycle owns `task_environments` and task-owned worktree lifetime. This +design makes hierarchy detachment, shared-workspace stewardship, creation-time +attachment, and destructive cleanup use one durable ownership contract. The +workspace system continues to own workspace and repository configuration; it +does not own task-environment cleanup authority. + +The design extends +[ADR-2026-08-08-task-owned-worktree-lifetime](../../../decisions/2026-08-08-task-owned-worktree-lifetime.md) +and applies the fail-closed rule from +[ADR-0009](../../../decisions/0009-fail-closed-gc-semantics.md). The ownership +generation and transaction rule is recorded in +[ADR-2026-09-04-generation-fenced-task-environment-ownership](../../../decisions/2026-09-04-generation-fenced-task-environment-ownership.md). + +## Requirement mapping + +| Requirement | Design sections | +| --- | --- | +| `REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-001` | [Ownership model](#ownership-model), [Atomic detachment](#atomic-detachment), [Cleanup fencing](#cleanup-fencing), [Restart and recovery](#restart-and-recovery) | +| `REQ-TASKS-DETACHED-WORKSPACE-CONTINUITY-002` | [Canonical creation attachment](#canonical-creation-attachment), [Concurrency and failure behavior](#concurrency-and-failure-behavior) | + +## Ownership model + +`task_workspace_groups.owner_task_id` identifies the current workspace steward. +Exactly one active membership row for that group has role `owner`; all other +active memberships have role `member`. The group gains an +`ownership_generation` integer that starts at `1` and increments whenever the +steward changes. + +`task_environments.task_id` remains the only physical environment owner. The +environment gains its own `ownership_generation`, also starting at `1` and +incremented by every successful owner transfer. A cleanup snapshot identifies +an environment by ID, owner task ID, and ownership generation. An environment +ID alone is never sufficient authority for a destructive operation. + +The workspace group can remain shared after detachment. Stewardship may move to +another active member later; this does not copy the workspace or invalidate +sessions that already reference the canonical environment. + +## Components and responsibilities + +- `internal/task/repository/sqlite` owns the cross-table transaction for + detachment, workspace-group stewardship, environment ownership generations, + and task cleanup barriers. It supports SQLite and PostgreSQL through the + existing dialect layer. +- `internal/task/service.Service.DetachTask` invokes only the transactional + repository operation, then reloads and publishes the committed task state. +- `internal/task/service.Service.CreateTask` owns required creation-time + workspace attachment before `task.created` is published or a launchable + result is returned. A narrow injected coordinator reuses + `HandoffService` policy resolution and membership behavior without making API + handlers responsible for correctness. +- `internal/task/service` task-resource cleanup validates environment ownership + and generation before teardown. Ownership transfers reject while the source + task has an active cleanup barrier. +- `internal/task/service.HandoffService` group cleanup claims the current group + generation before external cleanup and completes the status transition only + for that same generation. + +## Atomic detachment + +The repository exposes one detachment operation that performs these steps in a +single database transaction: + +1. Lock the child and former-parent task rows in deterministic task-ID order. +2. Reject when either task has a prepared, pending, running, or retry-wait + resource-cleanup job. +3. Lock the child's active workspace-group row and its materialized environment + row when present. +4. Clear `tasks.parent_id` and normalize `inherit_parent` to `shared_group`. +5. Set the group's steward and owner membership role to the child, demote the + previous owner membership to `member`, and increment the group generation. +6. Transfer the materialized environment to the child and increment the + environment generation. +7. Commit once. Any missing, ambiguous, archived, or concurrently changed row + rolls the complete operation back. + +Detaching an already-root task is an idempotent read of the current state. It +does not increment either generation. + +SQLite's writer transaction provides serialization. PostgreSQL uses explicit +row locks in the order above. Archive and delete preparation already reserve a +task cleanup barrier; detachment uses the same barrier states so either the +transfer commits before cleanup inventory is captured or detachment is rejected +after cleanup wins. + +## Canonical creation attachment + +Workspace policy resolution and attachment become part of the task service's +required synchronous create sequence. REST, WebSocket, MCP, plugin Host API, +workflow child creation, and other internal callers pass parent and explicit +workspace-policy inputs through `CreateTask`; no caller performs an independent +post-create membership write. + +The service resolves parent defaults before building task metadata. After the +task row is inserted, it records the workspace-group membership and sequential +blocker attachment before publishing `task.created` or returning a Created +outcome. If required attachment fails, the existing partial-task rollback path +removes the child and its dependent rows. Deduplicated external-ID outcomes do +not reapply workspace policy. + +HTTP and MCP handlers stop calling `AttachWorkspacePolicy` directly. The legacy +raw WebSocket create path and internal `CreateChildTask` path thereby gain the +same attachment behavior rather than bypassing it. + +## Cleanup fencing + +Task-resource cleanup snapshots persist the environment owner task ID and +ownership generation captured after the cleanup barrier is reserved. Before +environment or associated worktree teardown, the repository compares the +snapshot with the live environment. A missing row is idempotent success only +when the resource is already positively known absent. An owner or generation +mismatch makes the destructive portion a safe no-op and records that the +snapshot was superseded. + +Every environment-owner transfer uses a guarded repository method. It checks +the expected owner and generation, verifies that the source owner has no active +cleanup barrier, and increments the generation in the same transaction as the +owner change. The existing delete and cascade ownership-transfer paths use this +method; no unguarded `UPDATE task_environments SET task_id = ...` remains. + +Workspace-group cleanup changes from check-then-delete to a generation claim. +The claim atomically verifies the group generation, ownership flags, cleanup +policy, and absence of active members before setting `cleanup_pending` for that +generation. Stewardship transfer is allowed only from `active` or retryable +`cleanup_failed` state with no active cleanup claim. Completion and retry writes +use the claimed generation; stale completions cannot mark a replacement group +cleaned or failed. + +Filesystem and provider cleaners retain their existing managed-root and exact +resource-identity guards. Generation fencing is an additional authorization +condition, not a replacement for those checks. + +## Concurrency and failure behavior + +- Detach loses to an already-admitted parent or child cleanup and returns a + conflict without clearing `parent_id`. +- Cleanup inventory captured after detachment cannot select the former parent's + transferred environment because the environment owner has already changed. +- A delayed cleanup or ownership retry with an old generation is treated as + superseded and performs no destructive work. +- Concurrent sibling detaches serialize on the shared group. Each committed + transfer leaves one active steward and increments generations once. +- Creation returns success only after membership attachment. A concurrent + parent lifecycle barrier causes creation to roll back instead of publishing a + child that can launch without its workspace relationship. +- Repository or event-publication failures never compensate by deleting a + workspace whose current ownership cannot be proven. + +## Persistence and migration + +Add `ownership_generation INTEGER NOT NULL DEFAULT 1` to +`task_workspace_groups` and `task_environments`. Fresh schemas include the +columns and replayable migrations add them to existing SQLite and PostgreSQL +databases. Existing rows retain their current owner at generation `1`; no +filesystem mutation occurs during migration. + +All selects, inserts, model projections, test fixtures, and table-rebuild copy +lists that cover these tables include the new columns. Schema tests use the real +task/Office repository initialization order and cover fresh and replayed startup +for both dialects, following ADR-0027. + +## Restart and recovery + +Ownership generations live in the database and are included in durable cleanup +snapshots. On restart, cleanup workers re-read the current owner and generation +before acting. A snapshot prepared before a stewardship transfer cannot regain +authority after restart. The detached task's sessions continue resolving the +same materialized environment ID from the workspace group and their persisted +`task_environment_id` values. + +## Observability + +Structured logs identify the task, group, environment, expected owner and +generation, current owner and generation, and the resulting disposition for +ownership transfers and rejected cleanup. Logs do not include workspace file +contents, credentials, or restore secrets.