fix five concurrency/leak bugs (swarm, specialist, oauth, mcp, cron) (#265)

Surfaced by a full-tree audit; each ships a regression test where the
behavior is deterministically testable.

- swarm: deliver a handoff's mailbox note BEFORE registering the new task,
  so a mailbox failure can't leave a phantom pending task in the coordinator.
- specialist: put a foreground sub-agent child in its own process group and
  group-kill on cancel/timeout (+ WaitDelay backstop), so a build/server it
  forked dies with it instead of orphaning.
- oauth: serialize on-demand token refresh per key (keyed mutex) and re-load
  inside the lock, so parallel callers don't each spend the single-use
  refresh token (all but one would otherwise fail at the token endpoint).
- mcp: treat a "source already gone" rename during legacy-token migration as
  success, so a concurrent migrator doesn't wrongly drop an OAuth MCP server.
- cron: claim a due job atomically (advance NextRunAt under the per-job lock)
  before running it, so two schedulers can't double-fire the same job.

internal/cli/cron_claim_test.go

@@ -0,0 +1,41 @@
+package cli
+
+import (
+	"bytes"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Gitlawb/zero/internal/cron"
+)
+
+func TestFireJobClaimPreventsDoubleFire(t *testing.T) {
+	now := time.Date(2026, 6, 9, 9, 0, 0, 0, time.UTC)
+	nowFn := func() time.Time { return now }
+	store := cron.NewStore(cron.StoreOptions{RootDir: t.TempDir(), Now: nowFn})
+	due, _ := store.Add(cron.Job{Expr: "0 9 * * *", Prompt: "fire me", Status: cron.StatusActive, NextRunAt: now.Add(-time.Minute)})
+
+	fx := &fakeExec{}
+	// Two schedulers fire the same due job concurrently; the atomic claim must let
+	// exactly one through, so the job never double-fires (M9).
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			job, _ := store.Get(due.ID)
+			var out, errb bytes.Buffer
+			fireJob(store, nowFn, job, &out, &errb, fx.run)
+		}()
+	}
+	wg.Wait()
+
+	if len(fx.calls) != 1 {
+		t.Fatalf("claim must let exactly one scheduler fire the due job, got %d", len(fx.calls))
+	}
+	// And NextRunAt is advanced to the next slot (tomorrow 09:00), not left due.
+	d, _ := store.Get(due.ID)
+	if !d.NextRunAt.After(now) {
+		t.Fatalf("NextRunAt must advance past now, got %s", d.NextRunAt)
+	}
+}

internal/cli/cron_run.go

@@ -134,6 +134,24 @@ func reconcileOverdue(store *cron.Store, now func() time.Time, ids []string, std
 // previous fire has already returned before the next tick — no overlap.
 func fireJob(store *cron.Store, now func() time.Time, job cron.Job, stdout io.Writer, stderr io.Writer, exec execRunner) {
 	fired := now()
+	// Atomically claim this fire before running it so concurrent schedulers (or
+	// --once overlapping the forever loop) can't both execute the same due job: the
+	// winner advances NextRunAt under the per-job lock, and a loser sees the
+	// advanced time and skips. The post-exec advance recomputes the same
+	// NextRunAt (sched.Next(fired) is deterministic), so the claim serializes the
+	// fire decision without changing any persisted schedule state (M9).
+	claimed, claimErr := claimFire(store, fired, job.ID)
+	if claimErr != nil {
+		if errors.Is(claimErr, cron.ErrJobNotFound) {
+			return // removed before we could claim
+		}
+		fmt.Fprintf(stderr, "warning: could not claim job %s: %v\n", job.ID, claimErr)
+		return
+	}
+	if !claimed {
+		return // another scheduler already claimed this slot, or the job was paused
+	}
+
 	args := []string{"exec", "--output-format", "stream-json", "--session-title", "cron:" + job.ID}
 	if job.Cwd != "" {
 		args = append(args, "--cwd", job.Cwd)
@@ -215,6 +233,36 @@ func fireJob(store *cron.Store, now func() time.Time, job cron.Job, stdout io.Wr
 	fmt.Fprintf(stdout, "fired %s -> exit %d (next: %s)\n", job.ID, code, formatCronTime(persisted.NextRunAt))
 }
 
+// claimFire atomically claims a due job's fire under the per-job lock and reports
+// whether THIS caller should run it. It returns false only when the job was paused
+// externally or another scheduler already advanced NextRunAt past `fired` (so a
+// valid recurring job never double-fires). A valid due job has its NextRunAt
+// advanced here to claim the slot; an invalid or exhausted schedule is left for the
+// caller to fire-once and pause (preserving existing single-scheduler behavior).
+func claimFire(store *cron.Store, fired time.Time, id string) (bool, error) {
+	claimed := true
+	_, err := store.Mutate(id, func(current cron.Job, readErr error) (cron.Job, error) {
+		if readErr != nil {
+			return current, readErr
+		}
+		if current.Status != cron.StatusActive {
+			claimed = false // paused/other — not ours to fire
+			return current, nil
+		}
+		if current.NextRunAt.After(fired) {
+			claimed = false // another scheduler already advanced this slot
+			return current, nil
+		}
+		if sched, perr := cron.Parse(current.Expr); perr == nil {
+			if nxt := sched.Next(fired); !nxt.IsZero() {
+				current.NextRunAt = nxt // advance to claim; concurrent schedulers now skip
+			}
+		}
+		return current, nil
+	})
+	return claimed, err
+}
+
 func cronTruncate(s string, max int) string {
 	if len(s) <= max {
 		return s

internal/mcp/oauth_store.go

@@ -239,7 +239,13 @@ func (store *TokenStore) migrateLegacy(legacyPath string) error {
 			return err
 		}
 	}
-	return os.Rename(legacyPath, legacyPath+".migrated")
+	// A concurrent migrator may have already renamed the legacy file; treat a
+	// "source is gone" rename as success so the server isn't wrongly dropped on
+	// startup just because it lost the cleanup race (the tokens migrated fine) (M8).
+	if err := os.Rename(legacyPath, legacyPath+".migrated"); err != nil && !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	return nil
 }
 
 // mcpKey builds and validates the unified store key for an MCP server token.

internal/oauth/manager.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -34,6 +35,11 @@ type Manager struct {
 	// openBrowser is invoked with the authorization URL for loopback logins.
 	// Tests inject a function that drives the loopback redirect.
 	openBrowser func(authURL string) error
+	// refreshLocks serializes concurrent refreshes per key so parallel callers
+	// don't each spend the single-use refresh token; the loser reuses the rotated
+	// token. refreshMu guards the map (M7).
+	refreshMu    sync.Mutex
+	refreshLocks map[string]*sync.Mutex
 }
 
 // ManagerOptions configures a Manager.
@@ -308,6 +314,15 @@ func (m *Manager) Handle401(ctx context.Context, key string) (string, error) {
 }
 
 func (m *Manager) refreshAndSave(ctx context.Context, key string, cfg Config, current Token) (string, error) {
+	lock := m.keyLock(key)
+	lock.Lock()
+	defer lock.Unlock()
+	// Re-load inside the critical section: if a concurrent caller already refreshed
+	// (the access token changed), reuse it rather than spending the single-use
+	// refresh token a second time — the provider would reject the second use (M7).
+	if reloaded, err := m.loadToken(key); err == nil && reloaded.AccessToken != current.AccessToken {
+		return reloaded.AccessToken, nil
+	}
 	refreshed, err := Refresh(ctx, m.client, cfg, current, m.now)
 	if err != nil {
 		return "", err
@@ -318,6 +333,22 @@ func (m *Manager) refreshAndSave(ctx context.Context, key string, cfg Config, cu
 	return refreshed.AccessToken, nil
 }
 
+// keyLock returns the per-key refresh mutex, creating it on first use, so all
+// refreshes for one key are serialized.
+func (m *Manager) keyLock(key string) *sync.Mutex {
+	m.refreshMu.Lock()
+	defer m.refreshMu.Unlock()
+	if m.refreshLocks == nil {
+		m.refreshLocks = map[string]*sync.Mutex{}
+	}
+	lock, ok := m.refreshLocks[key]
+	if !ok {
+		lock = &sync.Mutex{}
+		m.refreshLocks[key] = lock
+	}
+	return lock
+}
+
 // loadToken loads a stored token for key without resolving any provider config,
 // so reading a still-valid login never depends on refresh-only configuration.
 func (m *Manager) loadToken(key string) (Token, error) {

internal/oauth/refresh_singleflight_test.go

@@ -0,0 +1,47 @@
+package oauth
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestRefreshSingleflightsConcurrentCallers(t *testing.T) {
+	// The fake token endpoint counts refreshes; with a single-use refresh token,
+	// only ONE concurrent refresh must happen — the rest reuse the rotated token (M7).
+	fp := newFakeProvider(t, `{"access_token":"fresh-at","expires_in":3600}`)
+	env := map[string]string{
+		"ZERO_OAUTH_DEMO_CLIENT_ID": "client",
+		"ZERO_OAUTH_DEMO_TOKEN_URL": fp.server.URL + "/token",
+	}
+	m := managerFor(t, env, nil)
+	if err := m.store.Save(ProviderKey("demo"), Token{AccessToken: "stale", RefreshToken: "rt", ExpiresAt: time.Now().Add(-time.Hour)}); err != nil {
+		t.Fatalf("seed: %v", err)
+	}
+
+	const N = 10
+	var wg sync.WaitGroup
+	got := make([]string, N)
+	errs := make([]error, N)
+	for i := 0; i < N; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			got[i], errs[i] = m.GetFresh(context.Background(), ProviderKey("demo"))
+		}(i)
+	}
+	wg.Wait()
+
+	for i := range got {
+		if errs[i] != nil {
+			t.Fatalf("GetFresh[%d]: %v", i, errs[i])
+		}
+		if got[i] != "fresh-at" {
+			t.Errorf("GetFresh[%d] = %q, want fresh-at", i, got[i])
+		}
+	}
+	if hits := fp.tokenHits.Load(); hits != 1 {
+		t.Fatalf("expected exactly 1 refresh across %d concurrent callers, got %d", N, hits)
+	}
+}

internal/specialist/exec.go

@@ -720,6 +720,9 @@ func cleanupPromptFile(promptFile string) {
 func runChildProcess(ctx context.Context, binaryPath string, args []string, progress func(streamjson.Event)) (ChildRunResult, error) {
 	var stderr bytes.Buffer
 	command := osexec.CommandContext(ctx, binaryPath, args...)
+	// Put the child in its own process group and group-kill on cancel/timeout, so a
+	// build/server/bash the sub-agent forked dies with it instead of orphaning (M6).
+	hardenSpecialistChild(command)
 	command.Stderr = &stderr
 	stdout, err := command.StdoutPipe()
 	if err != nil {

internal/specialist/exec_proc_unix.go

@@ -0,0 +1,40 @@
+//go:build !windows
+
+package specialist
+
+import (
+	"errors"
+	"os/exec"
+	"syscall"
+	"time"
+)
+
+// specialistWaitDelay bounds how long Wait blocks for the child's stdout pipe to
+// drain after the process exits or its context is cancelled, so a leaked
+// grandchild holding the pipe cannot hang the parent past cancel/timeout. Var
+// (not const) so tests can shorten it.
+var specialistWaitDelay = 2 * time.Second
+
+// hardenSpecialistChild makes a foreground specialist child killable as a single
+// unit. Setpgid puts the child into its own process group, so on cancel/timeout
+// we signal the whole group (negative pid) — any build/server/bash the sub-agent
+// forked dies with it instead of being orphaned. WaitDelay is the backstop if a
+// grandchild still holds the stdout pipe after the group is killed. Must be
+// called before command.Start (M6).
+func hardenSpecialistChild(command *exec.Cmd) {
+	if command.SysProcAttr == nil {
+		command.SysProcAttr = &syscall.SysProcAttr{}
+	}
+	command.SysProcAttr.Setpgid = true
+	command.WaitDelay = specialistWaitDelay
+	command.Cancel = func() error {
+		if command.Process == nil {
+			return nil
+		}
+		// Negative pid targets the whole process group led by the child.
+		if err := syscall.Kill(-command.Process.Pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
+			return err
+		}
+		return nil
+	}
+}

internal/specialist/exec_proc_unix_test.go

@@ -0,0 +1,22 @@
+//go:build !windows
+
+package specialist
+
+import (
+	"os/exec"
+	"testing"
+)
+
+func TestHardenSpecialistChild(t *testing.T) {
+	cmd := exec.Command("true")
+	hardenSpecialistChild(cmd)
+	if cmd.SysProcAttr == nil || !cmd.SysProcAttr.Setpgid {
+		t.Error("expected Setpgid (own process group) so cancel can group-kill grandchildren (M6)")
+	}
+	if cmd.WaitDelay == 0 {
+		t.Error("expected a WaitDelay backstop")
+	}
+	if cmd.Cancel == nil {
+		t.Error("expected a custom group-kill Cancel")
+	}
+}

internal/specialist/exec_proc_windows.go

@@ -0,0 +1,22 @@
+//go:build windows
+
+package specialist
+
+import (
+	"os/exec"
+	"time"
+)
+
+// specialistWaitDelay bounds how long Wait blocks for the child's stdout pipe to
+// drain after the process exits or its context is cancelled, so a leaked
+// grandchild holding the pipe cannot hang the parent past cancel/timeout. Var
+// (not const) so tests can shorten it.
+var specialistWaitDelay = 2 * time.Second
+
+// hardenSpecialistChild sets WaitDelay so a leaked grandchild cannot block Wait
+// indefinitely. Windows lacks POSIX process groups; tree-killing is not wired for
+// the synchronous specialist path, so the default Cancel (Process.Kill) plus
+// WaitDelay is used (M6).
+func hardenSpecialistChild(command *exec.Cmd) {
+	command.WaitDelay = specialistWaitDelay
+}

internal/swarm/lifecycle.go

@@ -110,13 +110,9 @@ func (s *Swarm) Handoff(pol Policy, teamName, taskID, toAgentType, note string)
 	if note != "" {
 		handoffTask += "\n\nHandoff note: " + note
 	}
-	if _, err := s.coord.Register(newID, newID, team, handoffTask); err != nil {
-		return "", err
-	}
-	cwd := s.cwdFor(taskID)
-	s.rememberCwd(newID, cwd)
-	// Deliver the handoff note to the new member's inbox. A mailbox failure must
-	// not silently strand the handoff, so it is surfaced to the caller.
+	// Deliver the handoff note to the new member's inbox BEFORE registering the new
+	// task, so a mailbox failure can't leave a phantom pending task in the
+	// coordinator (it returns the error having registered nothing) (M5).
 	if note != "" {
 		if mbErr := s.mailbox.Send(team, newID, Message{
 			From: task.AgentID, Subject: "handoff", Body: note, Type: "handoff", Time: nowRFC3339(),
@@ -124,6 +120,11 @@ func (s *Swarm) Handoff(pol Policy, teamName, taskID, toAgentType, note string)
 			return "", fmt.Errorf("swarm: deliver handoff note: %w", mbErr)
 		}
 	}
+	if _, err := s.coord.Register(newID, newID, team, handoffTask); err != nil {
+		return "", err
+	}
+	cwd := s.cwdFor(taskID)
+	s.rememberCwd(newID, cwd)
 	// Retire the original task (it has been re-delegated).
 	_ = s.coord.SetStatus(taskID, StatusHandedOff)
 	spec := s.buildSpec(pol, newID, newID, team, def, handoffTask, cwd)