@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 '',
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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 '',
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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{} {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 |
|
||||
|
||||
@@ -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.
|
||||
@@ -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='<isolated test 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`.
|
||||
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
@@ -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).
|
||||
See [the implementation plan](../../../plans/subtask-detachment/plan.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.
|
||||
Reference in New Issue
Block a user