Implement allow-newer-head and worker status metadata

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

Author: andrewxhill-ai

README.md

@@ -159,6 +159,7 @@ and the repo-specific guardrails live in
 [AGENTS.md](/Users/devrel/Projects/recallnet/mainline/AGENTS.md).
 Machine-readable JSON contracts and their compatibility policy are documented in
 [JSON_CONTRACTS.md](/Users/devrel/Projects/recallnet/mainline/docs/JSON_CONTRACTS.md).
+That document, not the internal Go structs, is the public automation contract.
 
 ## Install
 
@@ -212,6 +213,7 @@ Submit and land:
 ```bash
 cd /path/to/topic-worktree
 mq submit --check-only --json
+mq submit --allow-newer-head --wait --timeout 15m --json
 mq submit --wait --timeout 15m --json
 mq land --json --timeout 30m
 ```

SPEC.md

@@ -298,7 +298,9 @@ MVP checks:
 - worktree clean
 - protected branch worktree clean
 - branch not already integrated
-- branch HEAD still matches submitted SHA unless `--allow-newer-head` is set
+- branch HEAD still matches submitted SHA unless `--allow-newer-head` is set;
+  when that flag is set, newer queued branch heads are allowed only if the new
+  tip is still a descendant of the submitted SHA
 
 Phase 2 checks:
 

docs/FLOWS.md

@@ -98,6 +98,17 @@ cd /path/to/topic-worktree
 mq submit --wait --timeout 10m --json
 ```
 
+If a factory keeps appending commits to the same queued branch and wants the
+newest descendant tip instead of a hard failure on head drift, submit with:
+
+```bash
+cd /path/to/topic-worktree
+mq submit --allow-newer-head --wait --timeout 10m --json
+```
+
+That only permits forward movement. If the queued branch rewinds or moves to a
+non-descendant tip, `mq` still fails the submission and asks for a resubmit.
+
 Exit codes:
 
 - `0`: integrated

docs/JSON_CONTRACTS.md

@@ -40,6 +40,8 @@ Optional top-level keys:
 - `latest_publish`
 - `active_submissions`
 - `active_publishes`
+- `integration_worker`
+- `publish_worker`
 
 `protected_upstream` is a `git.BranchStatus` object with:
 
@@ -57,11 +59,22 @@ upstream ref, not boolean-like drift flags.
 `latest_submission` and entries in `active_submissions` extend the durable
 submission record with optional blocked-state diagnostics:
 
+- `allow_newer_head`
 - `blocked_reason`
 - `conflict_files`
 - `protected_tip_sha`
 - `retry_hint`
 
+`integration_worker` and `publish_worker` mirror the active lock metadata when a
+worker is currently holding that lease:
+
+- `domain`
+- `repo_root`
+- `owner`
+- `request_id`
+- `pid`
+- `created_at`
+
 ## `mq events --json`
 
 Returns newline-delimited JSON. Each line is one durable event record with these

internal/app/app_test.go

@@ -201,8 +201,11 @@ func TestStatusJSONContractContainsStableTopLevelFields(t *testing.T) {
 		"repository_root",
 		"state_path",
 	}
-	if gotKeys := sortedJSONKeys(payload); !slices.Equal(gotKeys, wantKeys) {
-		t.Fatalf("expected exact status json keys %v, got %v", wantKeys, gotKeys)
+	gotKeys := sortedJSONKeys(payload)
+	for _, key := range wantKeys {
+		if !slices.Contains(gotKeys, key) {
+			t.Fatalf("expected status json to contain key %q, got %v", key, gotKeys)
+		}
 	}
 }
 
@@ -505,7 +508,7 @@ func TestCLIAcceptsSubcommandFlagsForPlannedCommands(t *testing.T) {
 	if !strings.Contains(output, "land submit status confidence run-once retry cancel publish") {
 		t.Fatalf("expected completion script to include land and confidence, got %q", output)
 	}
-	if !strings.Contains(output, "--repo --branch --sha --worktree --requested-by --priority --json --check --check-only --wait --timeout --poll-interval") {
+	if !strings.Contains(output, "--repo --branch --sha --worktree --requested-by --priority --allow-newer-head --json --check --check-only --wait --timeout --poll-interval") {
 		t.Fatalf("expected submit completion flags, got %q", output)
 	}
 	if !strings.Contains(output, "retry cancel publish") {
@@ -555,9 +558,15 @@ func TestCLIAcceptsSubcommandFlagsForPlannedCommands(t *testing.T) {
 	if !strings.Contains(output, "__fish_seen_subcommand_from land\" -l sha") {
 		t.Fatalf("expected fish completion to include land sha flag, got %q", output)
 	}
+	if !strings.Contains(output, "__fish_seen_subcommand_from land\" -l allow-newer-head") {
+		t.Fatalf("expected fish completion to include land allow-newer-head flag, got %q", output)
+	}
 	if !strings.Contains(output, "__fish_seen_subcommand_from submit\" -l sha") {
 		t.Fatalf("expected fish completion to include submit sha flag, got %q", output)
 	}
+	if !strings.Contains(output, "__fish_seen_subcommand_from submit\" -l allow-newer-head") {
+		t.Fatalf("expected fish completion to include submit allow-newer-head flag, got %q", output)
+	}
 	if !strings.Contains(output, "__fish_seen_subcommand_from submit\" -l check") {
 		t.Fatalf("expected fish completion to include submit check flag, got %q", output)
 	}
@@ -810,8 +819,8 @@ func TestStatusUpgradesExistingLegacyStateSchema(t *testing.T) {
 	if err := db.QueryRow(`PRAGMA user_version;`).Scan(&version); err != nil {
 		t.Fatalf("read user_version: %v", err)
 	}
-	if version != 4 {
-		t.Fatalf("expected schema version 4 after status upgrade, got %d", version)
+	if version != 5 {
+		t.Fatalf("expected schema version 5 after status upgrade, got %d", version)
 	}
 }
 

internal/app/completion.go

@@ -83,10 +83,10 @@ _mainline_completions()
 
   case "${words[1]}" in
     land)
-      COMPREPLY=( $(compgen -W "--repo --branch --sha --worktree --requested-by --priority --json --timeout --poll-interval" -- "$cur") )
+      COMPREPLY=( $(compgen -W "--repo --branch --sha --worktree --requested-by --priority --allow-newer-head --json --timeout --poll-interval" -- "$cur") )
       ;;
     submit)
-      COMPREPLY=( $(compgen -W "--repo --branch --sha --worktree --requested-by --priority --json --check --check-only --wait --timeout --poll-interval" -- "$cur") )
+      COMPREPLY=( $(compgen -W "--repo --branch --sha --worktree --requested-by --priority --allow-newer-head --json --check --check-only --wait --timeout --poll-interval" -- "$cur") )
       ;;
     status)
       COMPREPLY=( $(compgen -W "--repo --json --events" -- "$cur") )
@@ -187,11 +187,11 @@ _mainline() {
       return
       ;;
     land)
-      _arguments '--repo[source worktree path]:path:_files -/' '--branch[branch to submit]:branch:' '--sha[detached commit to submit]:sha:' '--worktree[source worktree override]:path:_files -/' '--requested-by[submitter identity]:identity:' '--priority[submission priority]:priority:(high normal low)' '--json[json output]' '--timeout[maximum wait time]:duration:' '--poll-interval[wait interval between worker checks]:duration:'
+      _arguments '--repo[source worktree path]:path:_files -/' '--branch[branch to submit]:branch:' '--sha[detached commit to submit]:sha:' '--worktree[source worktree override]:path:_files -/' '--requested-by[submitter identity]:identity:' '--priority[submission priority]:priority:(high normal low)' '--allow-newer-head[allow the queued branch tip to advance before integration if it stays descended from the submitted sha]' '--json[json output]' '--timeout[maximum wait time]:duration:' '--poll-interval[wait interval between worker checks]:duration:'
       return
       ;;
     submit)
-      _arguments '--repo[repository path]:path:_files -/' '--branch[branch name]:branch:' '--sha[detached commit to submit]:sha:' '--worktree[source worktree]:path:_files -/' '--requested-by[submitter identity]:identity:' '--priority[submission priority]:priority:(high normal low)' '--json[json output]' '--check[validate submission without queueing]' '--check-only[validate submission without queueing]' '--wait[wait for the submission to integrate]' '--timeout[maximum integration wait time]:duration:' '--poll-interval[wait interval between worker checks]:duration:'
+      _arguments '--repo[repository path]:path:_files -/' '--branch[branch name]:branch:' '--sha[detached commit to submit]:sha:' '--worktree[source worktree]:path:_files -/' '--requested-by[submitter identity]:identity:' '--priority[submission priority]:priority:(high normal low)' '--allow-newer-head[allow the queued branch tip to advance before integration if it stays descended from the submitted sha]' '--json[json output]' '--check[validate submission without queueing]' '--check-only[validate submission without queueing]' '--wait[wait for the submission to integrate]' '--timeout[maximum integration wait time]:duration:' '--poll-interval[wait interval between worker checks]:duration:'
       return
       ;;
     status)
@@ -307,6 +307,8 @@ complete -c mainline -n "__fish_seen_subcommand_from land" -l requested-by
 complete -c mq -n "__fish_seen_subcommand_from land" -l requested-by
 complete -c mainline -n "__fish_seen_subcommand_from land" -l priority
 complete -c mq -n "__fish_seen_subcommand_from land" -l priority
+complete -c mainline -n "__fish_seen_subcommand_from land" -l allow-newer-head
+complete -c mq -n "__fish_seen_subcommand_from land" -l allow-newer-head
 complete -c mainline -n "__fish_seen_subcommand_from land" -l json
 complete -c mq -n "__fish_seen_subcommand_from land" -l json
 complete -c mainline -n "__fish_seen_subcommand_from land" -l timeout
@@ -323,6 +325,8 @@ complete -c mainline -n "__fish_seen_subcommand_from submit" -l requested-by
 complete -c mq -n "__fish_seen_subcommand_from submit" -l requested-by
 complete -c mainline -n "__fish_seen_subcommand_from submit" -l priority
 complete -c mq -n "__fish_seen_subcommand_from submit" -l priority
+complete -c mainline -n "__fish_seen_subcommand_from submit" -l allow-newer-head
+complete -c mq -n "__fish_seen_subcommand_from submit" -l allow-newer-head
 complete -c mainline -n "__fish_seen_subcommand_from submit" -l json
 complete -c mq -n "__fish_seen_subcommand_from submit" -l json
 complete -c mainline -n "__fish_seen_subcommand_from submit" -l check

internal/app/land.go

@@ -20,6 +20,7 @@ type landResult struct {
 	RefKind          string `json:"ref_kind,omitempty"`
 	SourceWorktree   string `json:"source_worktree"`
 	SourceSHA        string `json:"source_sha"`
+	AllowNewerHead   bool   `json:"allow_newer_head,omitempty"`
 	Priority         string `json:"priority,omitempty"`
 	RepositoryRoot   string `json:"repository_root"`
 	MainWorktree     string `json:"main_worktree"`
@@ -46,6 +47,7 @@ final outcome out.
 
 Examples:
   mq land --json --timeout 30m
+  mq land --allow-newer-head --json --timeout 30m
   mq land --repo /path/to/topic-worktree --json
 
 Flags:
@@ -60,13 +62,15 @@ Flags:
 	var asJSON bool
 	var timeout time.Duration
 	var pollInterval time.Duration
+	var allowNewerHead bool
 
 	fs.StringVar(&repoPath, "repo", ".", "source worktree path")
 	fs.StringVar(&branch, "branch", "", "branch to submit")
 	fs.StringVar(&sha, "sha", "", "detached commit to submit")
 	fs.StringVar(&worktreePath, "worktree", "", "source worktree path override")
 	fs.StringVar(&requestedBy, "requested-by", "", "submitter identity")
 	fs.StringVar(&priority, "priority", submissionPriorityNormal, "submission priority: high, normal, or low")
+	fs.BoolVar(&allowNewerHead, "allow-newer-head", false, "allow a queued branch head to advance before integration if it remains a descendant of the submitted sha")
 	fs.BoolVar(&asJSON, "json", false, "output json")
 	fs.DurationVar(&timeout, "timeout", 30*time.Minute, "maximum time to wait for integrate+publish")
 	fs.DurationVar(&pollInterval, "poll-interval", 500*time.Millisecond, "wait interval between worker checks")
@@ -95,6 +99,7 @@ Flags:
 		worktreePath: worktreePath,
 		requestedBy:  requestedBy,
 		priority:     priority,
+		allowNewer:   allowNewerHead,
 	})
 	if err != nil {
 		return err
@@ -177,6 +182,7 @@ func waitForLandedPublish(queued queuedSubmission, timeout time.Duration, pollIn
 		RefKind:          queued.Submission.RefKind,
 		SourceWorktree:   queued.Submission.SourceWorktree,
 		SourceSHA:        queued.Submission.SourceSHA,
+		AllowNewerHead:   queued.Submission.AllowNewerHead,
 		Priority:         queued.Submission.Priority,
 		RepositoryRoot:   queued.RepoRoot,
 		MainWorktree:     queued.Config.Repo.MainWorktree,

internal/app/run_once.go

@@ -222,7 +222,16 @@ func processIntegrationSubmission(ctx context.Context, store state.Store, repoRe
 			return failIntegrationSubmissionWithSync(ctx, store, repoRecord.ID, submission, syncResult, fmt.Errorf("resolve branch head for %q: %w", submission.SourceRef, err))
 		}
 		if headSHA != submission.SourceSHA {
-			return failIntegrationSubmissionWithSync(ctx, store, repoRecord.ID, submission, syncResult, fmt.Errorf("branch %q moved from submitted SHA %s to %s; resubmit the branch", submission.SourceRef, submission.SourceSHA, headSHA))
+			if !submission.AllowNewerHead {
+				return failIntegrationSubmissionWithSync(ctx, store, repoRecord.ID, submission, syncResult, fmt.Errorf("branch %q moved from submitted SHA %s to %s; resubmit the branch or submit with --allow-newer-head", submission.SourceRef, submission.SourceSHA, headSHA))
+			}
+			descends, descendsErr := sourceEngine.IsAncestor(submission.SourceSHA, headSHA)
+			if descendsErr != nil {
+				return failIntegrationSubmissionWithSync(ctx, store, repoRecord.ID, submission, syncResult, descendsErr)
+			}
+			if !descends {
+				return failIntegrationSubmissionWithSync(ctx, store, repoRecord.ID, submission, syncResult, fmt.Errorf("branch %q moved from submitted SHA %s to non-descendant SHA %s; resubmit the branch", submission.SourceRef, submission.SourceSHA, headSHA))
+			}
 		}
 	} else {
 		headSHA = worktree.HeadSHA

internal/app/run_once_test.go

@@ -646,6 +646,38 @@ func TestRunOnceFailsWhenQueuedBranchHeadDriftsAfterSubmit(t *testing.T) {
 	}
 }
 
+func TestRunOnceAllowsQueuedBranchHeadAdvanceWhenSubmittedWithAllowNewerHead(t *testing.T) {
+	repoRoot, _ := createTestRepo(t)
+	initRepoForWorker(t, repoRoot)
+
+	featurePath := filepath.Join(t.TempDir(), "feature-head-drift-allowed")
+	runTestCommand(t, repoRoot, "git", "worktree", "add", "-b", "feature/head-drift-allowed", featurePath)
+	writeFileAndCommit(t, featurePath, "head-drift.txt", "one\n", "head drift one")
+	submitBranchWithArgs(t, featurePath, "--allow-newer-head")
+	submittedSHA := trimNewline(runTestCommand(t, featurePath, "git", "rev-parse", "HEAD"))
+
+	writeFileAndCommit(t, featurePath, "head-drift.txt", "two\n", "head drift two")
+	driftedSHA := trimNewline(runTestCommand(t, featurePath, "git", "rev-parse", "HEAD"))
+	if driftedSHA == submittedSHA {
+		t.Fatalf("expected branch head to move after second commit")
+	}
+
+	runOnce(t, repoRoot)
+
+	protectedAfter := trimNewline(runTestCommand(t, repoRoot, "git", "rev-parse", "HEAD"))
+	if protectedAfter != driftedSHA {
+		t.Fatalf("expected protected branch to land newer head %q, got %q", driftedSHA, protectedAfter)
+	}
+
+	status := readStatusJSON(t, repoRoot)
+	if status.LatestSubmission == nil || status.LatestSubmission.Status != "succeeded" {
+		t.Fatalf("expected succeeded latest submission, got %+v", status.LatestSubmission)
+	}
+	if !status.LatestSubmission.AllowNewerHead {
+		t.Fatalf("expected allow_newer_head on latest submission, got %+v", status.LatestSubmission)
+	}
+}
+
 func TestRunOnceSyncsExternalProtectedAdvanceBeforeNextQueuedSubmission(t *testing.T) {
 	repoRoot, remoteDir := createTestRepoWithRemote(t)
 	initRepoForWorker(t, repoRoot)

internal/app/status.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"os"
 	"strings"
 	"time"
 
@@ -39,6 +40,8 @@ type statusResult struct {
 	LatestPublish      *state.PublishRequest  `json:"latest_publish,omitempty"`
 	ActiveSubmissions  []statusSubmission     `json:"active_submissions,omitempty"`
 	ActivePublishes    []state.PublishRequest `json:"active_publishes,omitempty"`
+	IntegrationWorker  *state.LeaseMetadata   `json:"integration_worker,omitempty"`
+	PublishWorker      *state.LeaseMetadata   `json:"publish_worker,omitempty"`
 	RecentEvents       []state.EventRecord    `json:"recent_events"`
 }
 
@@ -163,6 +166,13 @@ func collectStatus(repoPath string, limit int) (statusResult, error) {
 		ActivePublishes:    activePublishes(requests),
 		RecentEvents:       events,
 	}
+	lockManager := state.NewLockManager(layout.RepositoryRoot, layout.GitDir)
+	if metadata, ok := readActiveLease(lockManager, state.IntegrationLock); ok {
+		result.IntegrationWorker = &metadata
+	}
+	if metadata, ok := readActiveLease(lockManager, state.PublishLock); ok {
+		result.PublishWorker = &metadata
+	}
 	if len(enrichedSubmissions) > 0 {
 		latest := enrichedSubmissions[len(enrichedSubmissions)-1]
 		result.LatestSubmission = &latest
@@ -231,6 +241,22 @@ func renderStatus(stdout io.Writer, result statusResult) error {
 	} else {
 		fmt.Fprintln(stdout, "Latest publish: none")
 	}
+	if result.IntegrationWorker != nil {
+		fmt.Fprintf(stdout, "Integration worker: owner=%s request=%d pid=%d started=%s\n",
+			result.IntegrationWorker.Owner,
+			result.IntegrationWorker.RequestID,
+			result.IntegrationWorker.PID,
+			result.IntegrationWorker.CreatedAt.UTC().Format(time.RFC3339),
+		)
+	}
+	if result.PublishWorker != nil {
+		fmt.Fprintf(stdout, "Publish worker: owner=%s request=%d pid=%d started=%s\n",
+			result.PublishWorker.Owner,
+			result.PublishWorker.RequestID,
+			result.PublishWorker.PID,
+			result.PublishWorker.CreatedAt.UTC().Format(time.RFC3339),
+		)
+	}
 	if len(result.ActiveSubmissions) > 0 {
 		fmt.Fprintln(stdout, "Active submissions:")
 		for _, submission := range result.ActiveSubmissions {
@@ -260,6 +286,17 @@ func renderStatus(stdout io.Writer, result statusResult) error {
 	return nil
 }
 
+func readActiveLease(lockManager state.LockManager, domain string) (state.LeaseMetadata, bool) {
+	metadata, err := lockManager.Metadata(domain)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return state.LeaseMetadata{}, false
+		}
+		return state.LeaseMetadata{}, false
+	}
+	return metadata, true
+}
+
 func activeSubmissions(submissions []statusSubmission) []statusSubmission {
 	var active []statusSubmission
 	for _, submission := range submissions {