diff --git a/internal/cli/cron_claim_test.go b/internal/cli/cron_claim_test.go new file mode 100644 index 00000000..4108e633 --- /dev/null +++ b/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) + } +} diff --git a/internal/cli/cron_run.go b/internal/cli/cron_run.go index 95c99059..759eb95d 100644 --- a/internal/cli/cron_run.go +++ b/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 diff --git a/internal/mcp/oauth_store.go b/internal/mcp/oauth_store.go index 78a0f822..ab1f2d93 100644 --- a/internal/mcp/oauth_store.go +++ b/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. diff --git a/internal/oauth/manager.go b/internal/oauth/manager.go index 97a8cf0c..47ef736e 100644 --- a/internal/oauth/manager.go +++ b/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) { diff --git a/internal/oauth/refresh_singleflight_test.go b/internal/oauth/refresh_singleflight_test.go new file mode 100644 index 00000000..e26a8f8b --- /dev/null +++ b/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) + } +} diff --git a/internal/specialist/exec.go b/internal/specialist/exec.go index b6d00d06..6c474da0 100644 --- a/internal/specialist/exec.go +++ b/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 { diff --git a/internal/specialist/exec_proc_unix.go b/internal/specialist/exec_proc_unix.go new file mode 100644 index 00000000..b7bc9b12 --- /dev/null +++ b/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 + } +} diff --git a/internal/specialist/exec_proc_unix_test.go b/internal/specialist/exec_proc_unix_test.go new file mode 100644 index 00000000..f8aa101d --- /dev/null +++ b/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") + } +} diff --git a/internal/specialist/exec_proc_windows.go b/internal/specialist/exec_proc_windows.go new file mode 100644 index 00000000..a1f92f76 --- /dev/null +++ b/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 +} diff --git a/internal/swarm/lifecycle.go b/internal/swarm/lifecycle.go index c2931c28..da8dd300 100644 --- a/internal/swarm/lifecycle.go +++ b/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)