fix(task): stop superseded session worktrees claiming cutover slots (#3328)
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
+139
@@ -13,6 +13,7 @@ import (
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
dbutil "github.com/kandev/kandev/internal/db"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
)
|
||||
|
||||
func TestCutover_CanonicalRepoSupersedesDivergentFlatOwner(t *testing.T) {
|
||||
@@ -215,3 +216,141 @@ func TestCutover_DoesNotLogRolledBackFlatDemotion(t *testing.T) {
|
||||
t.Fatalf("rolled-back flat demotion logs = %d, want none", len(entries))
|
||||
}
|
||||
}
|
||||
|
||||
// TestCutover_SupersededHistoricalSessionWorktreeKeepsCanonicalSlot covers the
|
||||
// shape reported in issue #2505: a terminal session keeps an active legacy
|
||||
// worktree row on the empty branch slot while the surviving flat environment
|
||||
// and its canonical repository row own the same repository on a named slot.
|
||||
// The session row is superseded, so it must stay out of the normalized graph
|
||||
// instead of opening a second slot the worktree inventory check then rejects.
|
||||
func TestCutover_SupersededHistoricalSessionWorktreeKeepsCanonicalSlot(t *testing.T) {
|
||||
db := openLegacyDB(t)
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
seed := legacySeed{
|
||||
envID: "env-superseded-slot",
|
||||
taskID: "task-superseded-slot",
|
||||
repoID: "repo-superseded-slot",
|
||||
sessionID: "sess-superseded-slot",
|
||||
}
|
||||
seedLegacyTask(t, db, seed, now)
|
||||
seedLegacyFlatEnv(t, db, seed, "wt-canonical-slot", "/tasks/canonical-slot", "feature/canonical", now)
|
||||
if _, err := db.Exec(db.Rebind(`
|
||||
INSERT INTO task_environment_repos (
|
||||
id, task_environment_id, repository_id, branch_slug,
|
||||
worktree_id, worktree_path, worktree_branch, position,
|
||||
error_message, created_at, updated_at
|
||||
) VALUES (?, ?, ?, 'main', ?, ?, ?, 0, '', ?, ?)`),
|
||||
"env-repo-superseded", seed.envID, seed.repoID,
|
||||
"wt-canonical-slot", "/tasks/canonical-slot", "feature/canonical", now, now); err != nil {
|
||||
t.Fatalf("seed canonical repository row: %v", err)
|
||||
}
|
||||
seedLegacySessionWorktree(t, db, seed.sessionID, "wt-superseded", seed.repoID, "",
|
||||
"/tasks/superseded", "feature/superseded", "active", now)
|
||||
|
||||
core, logs := observer.New(zapcore.WarnLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
if err != nil {
|
||||
t.Fatalf("create observer logger: %v", err)
|
||||
}
|
||||
repo, err := NewWithDB(db, db, log)
|
||||
if err != nil {
|
||||
t.Fatalf("cutover: %v", err)
|
||||
}
|
||||
env, err := repo.GetTaskEnvironment(context.Background(), seed.envID)
|
||||
if err != nil {
|
||||
t.Fatalf("get task environment: %v", err)
|
||||
}
|
||||
if len(env.Repos) != 1 {
|
||||
t.Fatalf("normalized repos = %+v, want only the canonical slot", env.Repos)
|
||||
}
|
||||
if env.Repos[0].WorktreeID != "wt-canonical-slot" || env.Repos[0].BranchSlug != "main" {
|
||||
t.Fatalf("normalized repo = %+v, want wt-canonical-slot on slot main", env.Repos[0])
|
||||
}
|
||||
entries := logs.FilterMessage("cutover: duplicate legacy worktrees demoted to history").All()
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("session demotion logs = %d, want one", len(entries))
|
||||
}
|
||||
logged := fmt.Sprint(entries[0].ContextMap()["worktrees"])
|
||||
for _, want := range []string{"sess-superseded-slot", "wt-superseded", "/tasks/superseded", "feature/superseded", "wt-canonical-slot"} {
|
||||
if !strings.Contains(logged, want) {
|
||||
t.Fatalf("session demotion log %q must mention %q", logged, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutover_PreservesActiveSessionWhenFlatOwnerIsDeletedCanonical(t *testing.T) {
|
||||
db := openLegacyDB(t)
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
seed := legacySeed{
|
||||
envID: "env-deleted-canonical",
|
||||
taskID: "task-deleted-canonical",
|
||||
repoID: "repo-deleted-canonical",
|
||||
sessionID: "sess-deleted-canonical",
|
||||
}
|
||||
seedLegacyTask(t, db, seed, now)
|
||||
seedLegacyFlatEnv(t, db, seed, "wt-deleted-canonical", "/tasks/deleted-canonical", "feature/deleted", now)
|
||||
addLegacyRepoLifecycleColumns(t, db)
|
||||
if _, err := db.Exec(db.Rebind(`
|
||||
INSERT INTO task_environment_repos (
|
||||
id, task_environment_id, repository_id, branch_slug,
|
||||
worktree_id, worktree_path, worktree_branch, position,
|
||||
error_message, status, created_at, updated_at, merged_at, deleted_at
|
||||
) VALUES (?, ?, ?, 'main', ?, ?, ?, 0, '', 'deleted', ?, ?, NULL, ?)`),
|
||||
"env-repo-deleted-canonical", seed.envID, seed.repoID,
|
||||
"wt-deleted-canonical", "/tasks/deleted-canonical", "feature/deleted", now, now, now); err != nil {
|
||||
t.Fatalf("seed deleted canonical repository row: %v", err)
|
||||
}
|
||||
seedLegacySessionWorktree(t, db, seed.sessionID, "wt-active-session", seed.repoID, "",
|
||||
"/tasks/active-session", "feature/active", "active", now)
|
||||
|
||||
repo, err := NewWithDB(db, db, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("cutover: %v", err)
|
||||
}
|
||||
env, err := repo.GetTaskEnvironment(context.Background(), seed.envID)
|
||||
if err != nil {
|
||||
t.Fatalf("get task environment: %v", err)
|
||||
}
|
||||
slots := make(map[string]*models.TaskEnvironmentRepo, len(env.Repos))
|
||||
for _, envRepo := range env.Repos {
|
||||
slots[envRepo.RepositoryID+"\x00"+envRepo.BranchSlug] = envRepo
|
||||
}
|
||||
deleted := slots[seed.repoID+"\x00main"]
|
||||
active := slots[seed.repoID+"\x00"]
|
||||
if deleted == nil || deleted.WorktreeID != "wt-deleted-canonical" || deleted.Status != "deleted" {
|
||||
t.Fatalf("deleted canonical repository = %+v", deleted)
|
||||
}
|
||||
if active == nil || active.WorktreeID != "wt-active-session" || active.Status == "deleted" {
|
||||
t.Fatalf("active session repository = %+v, normalized repos = %+v", active, env.Repos)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCutover_PreservesDeletedSessionWorktreeHistory(t *testing.T) {
|
||||
db := openLegacyDB(t)
|
||||
now := time.Now().UTC().Truncate(time.Second)
|
||||
seed := legacySeed{
|
||||
envID: "env-deleted-session-history",
|
||||
taskID: "task-deleted-session-history",
|
||||
sessionID: "sess-deleted-session-history",
|
||||
}
|
||||
seedLegacyTask(t, db, seed, now)
|
||||
seedLegacyFlatEnv(t, db, legacySeed{envID: seed.envID, taskID: seed.taskID}, "", "", "", now)
|
||||
seedLegacySessionWorktree(t, db, seed.sessionID, "wt-deleted-session-history", "repo-deleted-session-history", "",
|
||||
"/tasks/deleted-session-history", "feature/deleted", "deleted", now)
|
||||
|
||||
repo, err := NewWithDB(db, db, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("cutover: %v", err)
|
||||
}
|
||||
env, err := repo.GetTaskEnvironment(context.Background(), seed.envID)
|
||||
if err != nil {
|
||||
t.Fatalf("get task environment: %v", err)
|
||||
}
|
||||
if len(env.Repos) != 1 {
|
||||
t.Fatalf("normalized repos = %+v, want deleted history", env.Repos)
|
||||
}
|
||||
deleted := env.Repos[0]
|
||||
if deleted.WorktreeID != "wt-deleted-session-history" || deleted.Status != "deleted" || deleted.DeletedAt == nil {
|
||||
t.Fatalf("deleted session worktree = %+v", deleted)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -564,13 +564,52 @@ func (c *worktreeCutover) mergeSessionWorktrees() {
|
||||
continue // another physical worktree owns this repository slot
|
||||
}
|
||||
targets := c.targetsForTask(c.sessionOwnerTaskID(wt.sessionID))
|
||||
if err := targets.mergeSessionWorktree(wt, c.isSupersededSessionWorktree(wt)); err != nil {
|
||||
historical := c.isSupersededSessionWorktree(wt)
|
||||
if historical && !isLegacyDeletedWorktree(wt) {
|
||||
if winner := c.demotionWinnerForSessionWorktree(wt, targets); winner != "" {
|
||||
c.demoteSessionWorktree(wt, winner)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if err := targets.mergeSessionWorktree(wt, historical); err != nil {
|
||||
c.conflicts = append(c.conflicts, fmt.Sprintf(
|
||||
"session %s worktree %s: %v", wt.sessionID, wt.worktreeID, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// demotionWinnerForSessionWorktree returns the active target that explains why
|
||||
// a historical session row is not an ownership source.
|
||||
func (c *worktreeCutover) demotionWinnerForSessionWorktree(
|
||||
wt legacySessionWorktree, targets *taskWorktreeTargets,
|
||||
) string {
|
||||
if targets.targetForWorktree(wt.worktreeID) != nil {
|
||||
return ""
|
||||
}
|
||||
if target, ok := targets.byKey[envRepoKey(wt.repositoryID, wt.branchSlug)]; ok &&
|
||||
target.worktreeID != "" && target.status != worktreeRepoStatusDeleted && target.deletedAt == nil {
|
||||
return target.worktreeID
|
||||
}
|
||||
session, ok := c.sessions[wt.sessionID]
|
||||
if !ok || !isLegacyHistoricalSession(session.state) {
|
||||
return ""
|
||||
}
|
||||
ownerTaskID := c.sessionOwnerTaskID(wt.sessionID)
|
||||
envID, ok := c.taskEnvIDs[ownerTaskID]
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
env, ok := c.envs[envID]
|
||||
if !ok || env.taskID != ownerTaskID || env.repositoryID != wt.repositoryID || env.worktreeID == "" {
|
||||
return ""
|
||||
}
|
||||
replacement := targets.activeTargetForWorktree(env.worktreeID)
|
||||
if replacement == nil || replacement.worktreeID == wt.worktreeID {
|
||||
return ""
|
||||
}
|
||||
return replacement.worktreeID
|
||||
}
|
||||
|
||||
func isLegacyDeletedWorktree(wt legacySessionWorktree) bool {
|
||||
return wt.status == worktreeRepoStatusDeleted || wt.deletedAt != nil
|
||||
}
|
||||
@@ -705,7 +744,7 @@ func (c *worktreeCutover) linkSessions() {
|
||||
// longer an ownership source. Rows demoted by the slot election, repository
|
||||
// rows, and the surviving flat environment take precedence for the same
|
||||
// physical identity regardless of session state. Historical rows with no
|
||||
// authoritative replacement remain eligible for backfill.
|
||||
// active authoritative replacement remain eligible for backfill.
|
||||
func (c *worktreeCutover) isSupersededSessionWorktree(wt legacySessionWorktree) bool {
|
||||
cacheKey := sessionWorktreeKey(wt.sessionID, wt.worktreeID)
|
||||
if c.demotedWorktrees[cacheKey] {
|
||||
@@ -731,7 +770,8 @@ func (c *worktreeCutover) isSupersededSessionWorktree(wt legacySessionWorktree)
|
||||
if !ok || env.taskID != ownerTaskID || row.worktreeID == "" {
|
||||
continue
|
||||
}
|
||||
if row.repositoryID == wt.repositoryID && row.branchSlug == wt.branchSlug {
|
||||
if row.repositoryID == wt.repositoryID && row.branchSlug == wt.branchSlug &&
|
||||
row.status != worktreeRepoStatusDeleted && row.deletedAt == nil {
|
||||
superseded = true
|
||||
break
|
||||
}
|
||||
@@ -757,7 +797,10 @@ func (c *worktreeCutover) flatEnvironmentSupersedesSessionWorktree(
|
||||
return false
|
||||
}
|
||||
env, ok := c.envs[envID]
|
||||
return ok && env.taskID == ownerTaskID && env.repositoryID == wt.repositoryID && env.worktreeID != ""
|
||||
if !ok || env.taskID != ownerTaskID || env.repositoryID != wt.repositoryID || env.worktreeID == "" {
|
||||
return false
|
||||
}
|
||||
return c.targetsForTask(ownerTaskID).activeTargetForWorktree(env.worktreeID) != nil
|
||||
}
|
||||
|
||||
// sessionOwnerTaskID returns the task that owns the environment a session
|
||||
|
||||
@@ -64,6 +64,16 @@ func (t *taskWorktreeTargets) targetForWorktree(worktreeID string) *envRepoTarge
|
||||
return nil
|
||||
}
|
||||
|
||||
// activeTargetForWorktree returns a target that can still own a physical
|
||||
// worktree during cutover. Deleted targets remain history, not ownership.
|
||||
func (t *taskWorktreeTargets) activeTargetForWorktree(worktreeID string) *envRepoTarget {
|
||||
target := t.targetForWorktree(worktreeID)
|
||||
if target == nil || target.status == worktreeRepoStatusDeleted || target.deletedAt != nil {
|
||||
return nil
|
||||
}
|
||||
return target
|
||||
}
|
||||
|
||||
// mergeLegacyEnvRepo merges a pre-existing environment-repository row. These
|
||||
// rows are the canonical source and win field conflicts (except for the
|
||||
// physical-worktree identity, which must agree everywhere).
|
||||
@@ -122,6 +132,15 @@ func (t *taskWorktreeTargets) mergeSessionWorktree(wt legacySessionWorktree, his
|
||||
}
|
||||
return existing.verifyWorktree(wt.worktreeID, wt.worktreePath, wt.worktreeBranch)
|
||||
}
|
||||
// A superseded row is historical evidence, never an ownership source, so
|
||||
// it must not open a repository slot of its own: the legacy worktree
|
||||
// inventory already excludes it, and claiming an unowned slot would add a
|
||||
// live worktree the inventory check then reports as extra. A deleted row
|
||||
// is the exception, because its claim carries the deleted status and stays
|
||||
// out of both inventories.
|
||||
if historical && !isLegacyDeletedWorktree(wt) {
|
||||
return nil
|
||||
}
|
||||
target := t.rowForKey(wt.repositoryID, wt.branchSlug)
|
||||
if historical && target.worktreeID != "" {
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user