Supersede stale blocked submissions by reachability

Co-authored-by: Andrew <andrewxhill@gmail.com>

Author: andrewxhill-ai

README.md

@@ -244,6 +244,8 @@ Recent hardening coverage now explicitly exercises the adoption-critical paths:
 
 - concurrent multi-worktree submit, integrate, and publish flows
 - deleted, moved, and dirtied source worktrees after submit
+- stale queued or blocked submissions whose source SHA is already reachable from
+  protected `main`, including after the original worktree was dropped
 - queued branch head drift after submit
 - external protected-branch advancement while queued work waits
 - inherited `pre-push` hook success and failure-plus-retry publish paths

SPEC.md

@@ -168,6 +168,9 @@ These are current design constraints, not hidden bugs:
   rebase and push
 - factories should currently integrate through the CLI/daemon contract, not by
   importing `internal/`
+- queued or blocked submissions whose `source_sha` is already reachable from
+  protected `main` are obsolete and may be auto-superseded even if their
+  original source worktree has been deleted
 
 ## Forward Product Direction
 

docs/FLOWS.md

@@ -281,6 +281,11 @@ without human branch surgery.
 `mq cancel --blocked` bulk-cancels blocked submissions when they are obsolete
 and you want to clear them out of the active queue surface.
 
+`mq run-once` and `mq doctor --fix` also auto-supersede queued or blocked
+submissions whose `source_sha` is already reachable from protected `main`, even
+if the original source worktree was deleted after that commit effectively
+landed.
+
 `mq status --json` now projects publish correlation back onto succeeded
 submissions through `publish_request_id`, `publish_status`, and `outcome`, so a
 factory can answer “did this submission fully land?” from one status surface.

internal/app/repo.go

@@ -1205,6 +1205,11 @@ func runDoctorFix(ctx context.Context, engine git.Engine, cfg policy.File, lockM
 			return nil, nil, err
 		}
 	}
+	if superseded, err := supersedeProtectedReachableSubmissions(ctx, store, repoRecord.ID, engine, cfg.Repo.ProtectedBranch); err != nil {
+		return nil, nil, err
+	} else if superseded > 0 {
+		applied = append(applied, fmt.Sprintf("superseded %d obsolete submission(s) already reachable from %s", superseded, cfg.Repo.ProtectedBranch))
+	}
 
 	if cfg.Repo.MainWorktree != "" && engine.BranchExists(cfg.Repo.ProtectedBranch) {
 		if _, err := os.Stat(cfg.Repo.MainWorktree); err == nil {

internal/app/repo_test.go

@@ -920,6 +920,61 @@ func TestDoctorJSONListsConcreteUnfinishedQueueItems(t *testing.T) {
 	}
 }
 
+func TestDoctorFixSupersedesBlockedSubmissionAlreadyReachableFromProtectedBranch(t *testing.T) {
+	repoRoot, _ := createTestRepoWithRemote(t)
+	initRepoForWorker(t, repoRoot)
+
+	featurePath := filepath.Join(t.TempDir(), "feature-doctor-reachable")
+	runTestCommand(t, repoRoot, "git", "worktree", "add", "-b", "feature/doctor-reachable", featurePath)
+	writeFileAndCommit(t, featurePath, "doctor.txt", "doctor\n", "doctor reachable feature")
+	sourceSHA := trimNewline(runTestCommand(t, featurePath, "git", "rev-parse", "HEAD"))
+	runTestCommand(t, repoRoot, "git", "merge", "--ff-only", "feature/doctor-reachable")
+
+	layout, err := git.DiscoverRepositoryLayout(repoRoot)
+	if err != nil {
+		t.Fatalf("DiscoverRepositoryLayout: %v", err)
+	}
+	store := state.NewStore(state.DefaultPath(layout.GitDir))
+	repoRecord, err := store.GetRepositoryByPath(context.Background(), layout.RepositoryRoot)
+	if err != nil {
+		t.Fatalf("GetRepositoryByPath: %v", err)
+	}
+	submission, err := store.CreateIntegrationSubmission(context.Background(), state.IntegrationSubmission{
+		RepoID:         repoRecord.ID,
+		BranchName:     "feature/doctor-reachable",
+		SourceRef:      "feature/doctor-reachable",
+		RefKind:        submissionRefKindBranch,
+		SourceWorktree: featurePath,
+		SourceSHA:      sourceSHA,
+		Status:         domain.SubmissionStatusBlocked,
+		LastError:      "waiting on dropped worktree",
+	})
+	if err != nil {
+		t.Fatalf("CreateIntegrationSubmission: %v", err)
+	}
+
+	if err := os.RemoveAll(featurePath); err != nil {
+		t.Fatalf("RemoveAll(featurePath): %v", err)
+	}
+
+	var out bytes.Buffer
+	var errOut bytes.Buffer
+	if err := runDoctor([]string{"--repo", repoRoot, "--fix", "--json"}, newStepPrinter(&out), &errOut); err != nil {
+		t.Fatalf("runDoctor --fix returned error: %v", err)
+	}
+
+	updated, err := store.GetIntegrationSubmission(context.Background(), submission.ID)
+	if err != nil {
+		t.Fatalf("GetIntegrationSubmission: %v", err)
+	}
+	if updated.Status != domain.SubmissionStatusSuperseded {
+		t.Fatalf("expected superseded blocked submission, got %+v", updated)
+	}
+	if !strings.Contains(updated.LastError, "already reachable from protected branch") {
+		t.Fatalf("expected reachability supersede reason, got %+v", updated)
+	}
+}
+
 func TestStatusJSONWorksWhenCanonicalRootIsDirty(t *testing.T) {
 	repoRoot, _ := createTestRepoWithRemote(t)
 	initRepoForWorker(t, repoRoot)

internal/app/run_once.go

@@ -110,6 +110,11 @@ func runOneCycle(repoPath string) (string, error) {
 		lease.Release()
 		return "", err
 	}
+	recoveredObsolete, err := supersedeProtectedReachableSubmissions(ctx, store, repoRecord.ID, git.NewEngine(mainLayout.WorktreeRoot), cfg.Repo.ProtectedBranch)
+	if err != nil {
+		lease.Release()
+		return "", err
+	}
 
 	submission, err := store.NextQueuedIntegrationSubmission(ctx, repoRecord.ID)
 	if err != nil {
@@ -159,6 +164,9 @@ func runOneCycle(repoPath string) (string, error) {
 	if err != nil {
 		return "", err
 	}
+	if result == "No queued publish requests." && recoveredObsolete > 0 {
+		return fmt.Sprintf("Superseded %d obsolete submission(s) already reachable from %s", recoveredObsolete, cfg.Repo.ProtectedBranch), nil
+	}
 	return result, nil
 }
 
@@ -791,6 +799,46 @@ func supersedeObsoleteSubmissions(ctx context.Context, store state.Store, repoID
 	return nil
 }
 
+func supersedeProtectedReachableSubmissions(ctx context.Context, store state.Store, repoID int64, engine git.Engine, protectedBranch string) (int, error) {
+	protectedSHA, err := engine.BranchHeadSHA(protectedBranch)
+	if err != nil {
+		return 0, err
+	}
+	submissions, err := store.ListIntegrationSubmissions(ctx, repoID)
+	if err != nil {
+		return 0, err
+	}
+	superseded := 0
+	for _, submission := range submissions {
+		if submission.Status != domain.SubmissionStatusQueued && submission.Status != domain.SubmissionStatusBlocked {
+			continue
+		}
+		if submission.SourceSHA == "" {
+			continue
+		}
+		reachable, err := engine.IsAncestor(submission.SourceSHA, protectedBranch)
+		if err != nil || !reachable {
+			continue
+		}
+		reason := fmt.Sprintf("superseded because source sha %s is already reachable from protected branch %q at %s", submission.SourceSHA, protectedBranch, protectedSHA)
+		if _, err := store.UpdateIntegrationSubmissionStatus(ctx, submission.ID, domain.SubmissionStatusSuperseded, reason); err != nil {
+			return superseded, err
+		}
+		if err := appendSubmissionEvent(ctx, store, repoID, submission.ID, domain.EventType("submission.superseded"), map[string]string{
+			"branch":               submissionDisplayRef(submission),
+			"source_ref":           submission.SourceRef,
+			"ref_kind":             string(submission.RefKind),
+			"source_sha":           submission.SourceSHA,
+			"superseded_protected": protectedSHA,
+			"reason":               reason,
+		}); err != nil {
+			return superseded, err
+		}
+		superseded++
+	}
+	return superseded, nil
+}
+
 func recoverInterruptedPublishRequests(ctx context.Context, store state.Store, repoID int64) (int, error) {
 	running, err := store.ListPublishRequestsByStatus(ctx, repoID, "running")
 	if err != nil {

internal/app/run_once_test.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/recallnet/mainline/internal/domain"
 	"github.com/recallnet/mainline/internal/git"
 	"github.com/recallnet/mainline/internal/policy"
 	"github.com/recallnet/mainline/internal/state"
@@ -750,6 +751,64 @@ func TestRunOnceFailsWhenSubmittedSourceWorktreeTurnsDirtyAfterSubmit(t *testing
 	}
 }
 
+func TestRunOnceSupersedesBlockedSubmissionAlreadyReachableAfterWorktreeDrop(t *testing.T) {
+	repoRoot, _ := createTestRepo(t)
+	initRepoForWorker(t, repoRoot)
+
+	featurePath := filepath.Join(t.TempDir(), "feature-reachable-dropped")
+	runTestCommand(t, repoRoot, "git", "worktree", "add", "-b", "feature/reachable-dropped", featurePath)
+	writeFileAndCommit(t, featurePath, "reachable.txt", "reachable\n", "reachable feature")
+	sourceSHA := trimNewline(runTestCommand(t, featurePath, "git", "rev-parse", "HEAD"))
+	runTestCommand(t, repoRoot, "git", "merge", "--ff-only", "feature/reachable-dropped")
+
+	layout, err := git.DiscoverRepositoryLayout(repoRoot)
+	if err != nil {
+		t.Fatalf("DiscoverRepositoryLayout: %v", err)
+	}
+	store := state.NewStore(state.DefaultPath(layout.GitDir))
+	repoRecord, err := store.GetRepositoryByPath(context.Background(), layout.RepositoryRoot)
+	if err != nil {
+		t.Fatalf("GetRepositoryByPath: %v", err)
+	}
+	submission, err := store.CreateIntegrationSubmission(context.Background(), state.IntegrationSubmission{
+		RepoID:         repoRecord.ID,
+		BranchName:     "feature/reachable-dropped",
+		SourceRef:      "feature/reachable-dropped",
+		RefKind:        submissionRefKindBranch,
+		SourceWorktree: featurePath,
+		SourceSHA:      sourceSHA,
+		Status:         domain.SubmissionStatusBlocked,
+		LastError:      "rebase conflict",
+	})
+	if err != nil {
+		t.Fatalf("CreateIntegrationSubmission: %v", err)
+	}
+
+	if err := os.RemoveAll(featurePath); err != nil {
+		t.Fatalf("RemoveAll(featurePath): %v", err)
+	}
+
+	var runOut bytes.Buffer
+	var runErr bytes.Buffer
+	if err := runRunOnce([]string{"--repo", repoRoot}, newStepPrinter(&runOut), &runErr); err != nil {
+		t.Fatalf("runRunOnce returned error: %v", err)
+	}
+	if !strings.Contains(runOut.String(), "Superseded 1 obsolete submission") {
+		t.Fatalf("expected supersede summary, got %q", runOut.String())
+	}
+
+	updated, err := store.GetIntegrationSubmission(context.Background(), submission.ID)
+	if err != nil {
+		t.Fatalf("GetIntegrationSubmission: %v", err)
+	}
+	if updated.Status != domain.SubmissionStatusSuperseded {
+		t.Fatalf("expected superseded blocked submission, got %+v", updated)
+	}
+	if !strings.Contains(updated.LastError, "already reachable from protected branch") {
+		t.Fatalf("expected reachability supersede reason, got %+v", updated)
+	}
+}
+
 func TestRunOnceFailsWhenQueuedBranchHeadDriftsAfterSubmit(t *testing.T) {
 	repoRoot, _ := createTestRepo(t)
 	initRepoForWorker(t, repoRoot)