diff --git a/CLI_AND_DAEMON.md b/CLI_AND_DAEMON.md index 62a1edaa7..ec321ed7a 100644 --- a/CLI_AND_DAEMON.md +++ b/CLI_AND_DAEMON.md @@ -220,6 +220,7 @@ Daemon behavior is configured via flags or environment variables: | GC artifact TTL (open issues) | — | `MULTICA_GC_ARTIFACT_TTL` | `12h` (set `0` to disable) | | GC artifact patterns | — | `MULTICA_GC_ARTIFACT_PATTERNS` | `node_modules,.next,.turbo` | | GC repo cache TTL (`.repos`) | — | `MULTICA_GC_REPO_TTL` | `720h` (30d; set `0` to disable) | +| GC repo maintenance | — | `MULTICA_GC_REPO_MAINTENANCE_ENABLED` | `true` (set `false`/`0` to disable heavy Git maintenance only) | | GC Hermes memory TTL (per-agent `memories/`) | — | `MULTICA_GC_HERMES_MEMORY_TTL` | `2160h` (90d; set `0` to disable) | #### Workspace garbage collection @@ -232,6 +233,8 @@ The daemon periodically scans `MULTICA_WORKSPACES_ROOT` and reclaims disk space - **Repo cache eviction** — the bare git clones under `.repos/` are shared object stores: each task workdir is a `git worktree` off one of them rather than its own clone, so a task's `.git` is only a pointer file. They are evicted only when all of the following hold: the repo is no longer attached to any workspace this daemon watches, it has no worktrees left, and no task has created a worktree from it for `MULTICA_GC_REPO_TTL`. A cache created before this stamp existed is not treated as ancient — its clock starts at the first GC cycle that sees it, so upgrading does not wipe every cache. Evicting is safe by construction: the next task that needs the repo re-clones it on demand, so a wrong eviction costs a clone, not a failure. + Short worktree cleanup and eligible cache eviction continue on every GC cycle, including while agents are active. Heavy repo maintenance (`reflog expire` and `git gc`) starts only while the daemon is otherwise idle. A checkout or newly claimed task cancels it and takes priority; interrupted work remains pending for a later idle GC cycle. Operators can disable only these heavy commands with `MULTICA_GC_REPO_MAINTENANCE_ENABLED=false` without disabling worktree cleanup or cache eviction. + - **Hermes memory store reclamation** — a Hermes agent's long-term memory (`memories/`) lives at `/hermes-state///`, outside any task directory, so it survives across tasks and issues (see [Hermes agent memory](#hermes-agent-memory)). A store untouched for `MULTICA_GC_HERMES_MEMORY_TTL` is removed, giving a deleted agent's memory an eventual-reclamation guarantee. The default is deliberately long: these are a handful of markdown files, and reclaiming one is user-visible amnesia rather than a cache miss. A store a running task holds is never reclaimed. Configured patterns are basename-only — entries containing `/` or `\` are silently dropped — and `.git` subtrees are never descended into. The managed Codex cache is matched by its exact relative path, so a repository's own `.sandbox-bin` is not removed unless an operator explicitly adds that basename to `MULTICA_GC_ARTIFACT_PATTERNS`. The default list (`node_modules`, `.next`, `.turbo`) is intentionally narrow; extend it per deployment if your repos consistently produce other regenerable directories (for example, `MULTICA_GC_ARTIFACT_PATTERNS=node_modules,.next,.turbo,target,__pycache__`). To disable artifact cleanup entirely, including the managed Codex cache, set `MULTICA_GC_ARTIFACT_TTL=0`. diff --git a/server/cmd/multica/cmd_repo.go b/server/cmd/multica/cmd_repo.go index e1907273c..7f266d534 100644 --- a/server/cmd/multica/cmd_repo.go +++ b/server/cmd/multica/cmd_repo.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "os" + "strconv" "strings" "time" @@ -349,7 +350,7 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error { return fmt.Errorf("get working directory: %w", err) } - reqBody := map[string]string{ + reqBody := map[string]any{ "url": repoURL, "workspace_id": workspaceID, "workdir": workDir, @@ -357,6 +358,7 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error { "agent_name": agentName, "task_id": taskID, "checkout_mode": strings.TrimSpace(os.Getenv("MULTICA_REPO_CHECKOUT_MODE")), + "retry_busy": true, } data, err := json.Marshal(reqBody) @@ -364,21 +366,48 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error { return fmt.Errorf("encode request: %w", err) } - client := &http.Client{Timeout: 5 * time.Minute} - resp, err := client.Post( - fmt.Sprintf("http://127.0.0.1:%s/repo/checkout", daemonPort), - "application/json", - bytes.NewReader(data), - ) - if err != nil { - return fmt.Errorf("connect to daemon: %w", err) + parentCtx := cmd.Context() + if parentCtx == nil { + parentCtx = context.Background() } - defer resp.Body.Close() - - body, _ := io.ReadAll(resp.Body) - - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("checkout failed: %s", string(body)) + ctx, cancel := context.WithTimeout(parentCtx, 5*time.Minute) + defer cancel() + client := &http.Client{} + checkoutURL := fmt.Sprintf("http://127.0.0.1:%s/repo/checkout", daemonPort) + var body []byte + for { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, checkoutURL, bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("create daemon checkout request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("connect to daemon: %w", err) + } + body, err = io.ReadAll(resp.Body) + closeErr := resp.Body.Close() + if err != nil { + return fmt.Errorf("read daemon checkout response: %w", err) + } + if closeErr != nil { + return fmt.Errorf("close daemon checkout response: %w", closeErr) + } + if resp.StatusCode == http.StatusServiceUnavailable && resp.Header.Get("X-Multica-Retryable") == "repo-busy" { + delay := repoCheckoutRetryDelay(resp.Header.Get("Retry-After"), time.Now()) + timer := time.NewTimer(delay) + select { + case <-ctx.Done(): + timer.Stop() + return fmt.Errorf("connect to daemon: %w", context.Cause(ctx)) + case <-timer.C: + continue + } + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("checkout failed: %s", string(body)) + } + break } var result struct { @@ -394,3 +423,18 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error { return nil } + +func repoCheckoutRetryDelay(value string, now time.Time) time.Duration { + const ( + defaultDelay = time.Second + maxDelay = 30 * time.Second + ) + value = strings.TrimSpace(value) + if seconds, err := strconv.Atoi(value); err == nil && seconds >= 0 { + return min(time.Duration(seconds)*time.Second, maxDelay) + } + if retryAt, err := http.ParseTime(value); err == nil { + return min(max(retryAt.Sub(now), time.Duration(0)), maxDelay) + } + return defaultDelay +} diff --git a/server/cmd/multica/cmd_repo_test.go b/server/cmd/multica/cmd_repo_test.go index 8a77f3829..07dd9030c 100644 --- a/server/cmd/multica/cmd_repo_test.go +++ b/server/cmd/multica/cmd_repo_test.go @@ -6,6 +6,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" "github.com/spf13/cobra" ) @@ -190,7 +191,7 @@ func TestRunRepoRemoveRejectsMissingRepoWithoutPatch(t *testing.T) { } func TestRunRepoCheckoutForwardsManagedCheckoutMode(t *testing.T) { - var body map[string]string + var body map[string]any srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost || r.URL.Path != "/repo/checkout" { http.NotFound(w, r) @@ -225,4 +226,68 @@ func TestRunRepoCheckoutForwardsManagedCheckoutMode(t *testing.T) { if got := body["ref"]; got != "release/v2" { t.Fatalf("ref = %q, want release/v2", got) } + if got := body["retry_busy"]; got != true { + t.Fatalf("retry_busy = %v, want true", got) + } +} + +func TestRunRepoCheckoutRetriesServiceUnavailable(t *testing.T) { + attempts := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts++ + if attempts == 1 { + w.Header().Set("X-Multica-Retryable", "repo-busy") + w.Header().Set("Retry-After", "0") + http.Error(w, "repository busy", http.StatusServiceUnavailable) + return + } + json.NewEncoder(w).Encode(map[string]string{ + "path": "/work/repo", + "branch_name": "agent/test/task", + }) + })) + defer srv.Close() + + t.Setenv("MULTICA_DAEMON_PORT", strings.TrimPrefix(srv.URL, "http://127.0.0.1:")) + t.Setenv("MULTICA_WORKSPACE_ID", "ws-1") + t.Setenv("MULTICA_AGENT_NAME", "Test Agent") + t.Setenv("MULTICA_TASK_ID", "task-1") + + if err := runRepoCheckout(&cobra.Command{}, []string{"https://github.com/org/repo.git"}); err != nil { + t.Fatalf("runRepoCheckout: %v", err) + } + if attempts != 2 { + t.Fatalf("checkout attempts = %d, want 2", attempts) + } +} + +func TestRunRepoCheckoutDoesNotRetryUnmarkedServiceUnavailable(t *testing.T) { + attempts := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + attempts++ + http.Error(w, "daemon unavailable", http.StatusServiceUnavailable) + })) + defer srv.Close() + + t.Setenv("MULTICA_DAEMON_PORT", strings.TrimPrefix(srv.URL, "http://127.0.0.1:")) + if err := runRepoCheckout(&cobra.Command{}, []string{"https://github.com/org/repo.git"}); err == nil { + t.Fatal("runRepoCheckout unexpectedly succeeded") + } + if attempts != 1 { + t.Fatalf("checkout attempts = %d, want 1", attempts) + } +} + +func TestRepoCheckoutRetryDelay(t *testing.T) { + t.Parallel() + now := time.Date(2026, 8, 12, 0, 0, 0, 0, time.UTC) + if got := repoCheckoutRetryDelay("7", now); got != 7*time.Second { + t.Fatalf("seconds delay = %s, want 7s", got) + } + if got := repoCheckoutRetryDelay(now.Add(time.Minute).Format(http.TimeFormat), now); got != 30*time.Second { + t.Fatalf("capped date delay = %s, want 30s", got) + } + if got := repoCheckoutRetryDelay("invalid", now); got != time.Second { + t.Fatalf("default delay = %s, want 1s", got) + } } diff --git a/server/internal/daemon/config.go b/server/internal/daemon/config.go index d94dd1aab..b5114326e 100644 --- a/server/internal/daemon/config.go +++ b/server/internal/daemon/config.go @@ -105,6 +105,7 @@ type Config struct { GCArtifactTTL time.Duration // when a task has been completed for at least this long but its issue is still open, drop regenerable artifacts (default: 12h, set 0 to disable) GCArtifactPatterns []string // basename patterns whose subtrees are removed during artifact cleanup (default: node_modules, .next, .turbo) GCRepoTTL time.Duration // evict a cached bare repo under .repos once no task has created a worktree from it for this long, it has no worktrees left, and it is no longer attached to any watched workspace (default: 30d, set 0 to disable) + GCRepoMaintenanceEnabled bool // run reflog expiry and git gc after stale agent refs are removed (default: true; disable independently as an operational kill switch) GCCodexSessionTTL time.Duration // reclaim a per-issue Codex session store (~/.codex/multica-sessions//) untouched for at least this long, so a done/abandoned issue's conversation history does not accumulate forever (default: 14d, set 0 to disable) GCHermesMemoryTTL time.Duration // reclaim a per-agent Hermes memory store (/hermes-state//) untouched for at least this long, so a deleted agent's memory does not sit on disk forever (default: 90d, set 0 to disable) AutoUpdateEnabled bool // periodically check for a newer CLI release and self-update when idle (default: true on Multica Cloud, false on self-host) @@ -430,6 +431,7 @@ func LoadConfig(overrides Overrides) (Config, error) { if err != nil { return Config{}, err } + gcRepoMaintenanceEnabled := boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true) gcArtifactPatterns := patternsFromEnv("MULTICA_GC_ARTIFACT_PATTERNS", DefaultGCArtifactPatterns) // Auto-update config: default -> env override -> CLI override. @@ -481,6 +483,7 @@ func LoadConfig(overrides Overrides) (Config, error) { GCArtifactTTL: gcArtifactTTL, GCArtifactPatterns: gcArtifactPatterns, GCRepoTTL: gcRepoTTL, + GCRepoMaintenanceEnabled: gcRepoMaintenanceEnabled, GCCodexSessionTTL: gcCodexSessionTTL, GCHermesMemoryTTL: gcHermesMemoryTTL, AutoUpdateEnabled: autoUpdateEnabled, diff --git a/server/internal/daemon/config_test.go b/server/internal/daemon/config_test.go index 6e8275229..0602cfc35 100644 --- a/server/internal/daemon/config_test.go +++ b/server/internal/daemon/config_test.go @@ -35,6 +35,18 @@ func TestDefaultGCIntervalIsTwoHours(t *testing.T) { } } +func TestRepoMaintenanceKillSwitchDefaultsOnAndCanDisable(t *testing.T) { + t.Setenv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", "") + if !boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true) { + t.Fatal("repo maintenance kill switch should default to enabled") + } + + t.Setenv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", "false") + if boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true) { + t.Fatal("repo maintenance kill switch should accept false") + } +} + func TestPatternsFromEnv_DropsSeparatorBearingEntries(t *testing.T) { t.Setenv("MULTICA_GC_ARTIFACT_PATTERNS", "node_modules, .next ,foo/bar, ../etc, ,target") got := patternsFromEnv("MULTICA_GC_ARTIFACT_PATTERNS", nil) diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 45f0e3490..051161238 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -272,7 +272,7 @@ type workspaceState struct { taskRepoRefs map[string]map[string]string // taskID -> repo URL -> checkout ref settings json.RawMessage // workspace settings (JSONB) lastRepoSyncErr string - repoRefreshMu sync.Mutex + repoRefreshMu contextLock // profileSetSig is a content hash of the workspace's custom runtime // profile list (MUL-3332) as last seen from the server. An on-demand // refresh compares the live signature with this cached value; any drift @@ -297,6 +297,38 @@ type workspaceState struct { builtinVersions map[string]string } +// contextLock is a zero-value-ready mutex whose wait can be cancelled. Repo +// checkout requests use it for workspace refresh coalescing so disconnecting a +// client never leaves the handler stuck behind another cold-cache refresh. +type contextLock struct { + once sync.Once + token chan struct{} +} + +func (l *contextLock) Lock(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return context.Cause(ctx) + } + l.once.Do(func() { + l.token = make(chan struct{}, 1) + l.token <- struct{}{} + }) + select { + case <-ctx.Done(): + return context.Cause(ctx) + case <-l.token: + if err := ctx.Err(); err != nil { + l.token <- struct{}{} + return context.Cause(ctx) + } + return nil + } +} + +func (l *contextLock) Unlock() { + l.token <- struct{}{} +} + type repoCacheBackend interface { Lookup(workspaceID, url string) string BarePath(workspaceID, url string) string @@ -2776,10 +2808,22 @@ func (d *Daemon) waitBackgroundSyncs() { } func (d *Daemon) syncWorkspaceRepos(workspaceID string, repos []RepoData) { + d.syncWorkspaceReposContext(context.Background(), workspaceID, repos) +} + +func (d *Daemon) syncWorkspaceReposContext(ctx context.Context, workspaceID string, repos []RepoData) { if d.repoCache == nil { return } - if err := d.repoCache.Sync(workspaceID, repoDataToInfo(repos)); err != nil { + var err error + if cache, ok := d.repoCache.(interface { + SyncContext(context.Context, string, []repocache.RepoInfo) error + }); ok { + err = cache.SyncContext(ctx, workspaceID, repoDataToInfo(repos)) + } else { + err = d.repoCache.Sync(workspaceID, repoDataToInfo(repos)) + } + if err != nil { d.setWorkspaceRepoSyncError(workspaceID, err.Error()) d.logger.Warn("repo cache sync failed", "workspace_id", workspaceID, "error", err) return @@ -3026,7 +3070,9 @@ func (d *Daemon) ensureRepoReady(ctx context.Context, workspaceID, repoURL strin // sibling's refresh is fresh enough for our gate read. cacheHitOnEntry := d.workspaceRepoAllowed(workspaceID, repoURL) && d.repoCache.Lookup(workspaceID, repoURL) != "" - ws.repoRefreshMu.Lock() + if err := ws.repoRefreshMu.Lock(ctx); err != nil { + return err + } defer ws.repoRefreshMu.Unlock() if !cacheHitOnEntry && d.workspaceRepoAllowed(workspaceID, repoURL) && d.repoCache.Lookup(workspaceID, repoURL) != "" { @@ -3046,7 +3092,10 @@ func (d *Daemon) ensureRepoReady(ctx context.Context, workspaceID, repoURL strin return nil } - d.syncWorkspaceRepos(workspaceID, resp.Repos) + d.syncWorkspaceReposContext(ctx, workspaceID, resp.Repos) + if err := ctx.Err(); err != nil { + return context.Cause(ctx) + } if d.repoCache.Lookup(workspaceID, repoURL) != "" { return nil @@ -4424,6 +4473,13 @@ func (d *Daemon) runBatchPoller(pollerCtx, parentCtx context.Context, sem chan i d.logger.Info("task received", "task", shortID(t.ID), "target", taskTarget) taskWG.Add(1) d.activeTasks.Add(1) + if cache, ok := d.repoCache.(interface{ CancelMaintenance() }); ok { + // A task can reuse an existing worktree and never enter the + // checkout path that normally preempts repository maintenance. + // Cancel all low-priority maintenance before the agent starts so + // direct Git operations cannot overlap it. + cache.CancelMaintenance() + } go func(t Task, slot int) { defer taskWG.Done() defer d.activeTasks.Add(-1) diff --git a/server/internal/daemon/daemon_test.go b/server/internal/daemon/daemon_test.go index 3fd5ab812..fdf84cceb 100644 --- a/server/internal/daemon/daemon_test.go +++ b/server/internal/daemon/daemon_test.go @@ -3794,6 +3794,26 @@ func TestEnsureRepoReadyConcurrentMissRefreshesOnce(t *testing.T) { } } +func TestContextLockCancelsWaitWithoutConsumingToken(t *testing.T) { + t.Parallel() + + var lock contextLock + if err := lock.Lock(context.Background()); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if err := lock.Lock(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Lock error = %v, want deadline exceeded", err) + } + lock.Unlock() + + if err := lock.Lock(context.Background()); err != nil { + t.Fatalf("Lock after cancelled waiter: %v", err) + } + lock.Unlock() +} + func TestShellArgsFromEnv(t *testing.T) { t.Setenv("MULTICA_CLAUDE_ARGS", `--max-turns 60 --append-system-prompt "multi word"`) got, err := shellArgsFromEnv("MULTICA_CLAUDE_ARGS") diff --git a/server/internal/daemon/gc.go b/server/internal/daemon/gc.go index 588d27f47..470816da8 100644 --- a/server/internal/daemon/gc.go +++ b/server/internal/daemon/gc.go @@ -11,6 +11,7 @@ import ( "time" "github.com/multica-ai/multica/server/internal/daemon/execenv" + "github.com/multica-ai/multica/server/internal/daemon/processtree" "github.com/multica-ai/multica/server/internal/daemon/repocache" ) @@ -32,6 +33,7 @@ func (d *Daemon) gcLoop(ctx context.Context) { "orphan_ttl", d.cfg.GCOrphanTTL, "artifact_ttl", d.cfg.GCArtifactTTL, "repo_ttl", d.cfg.GCRepoTTL, + "repo_maintenance_enabled", d.cfg.GCRepoMaintenanceEnabled, "artifact_patterns", d.cfg.GCArtifactPatterns, "managed_artifact_subpaths", execenv.ManagedReclaimableArtifactSubpaths(), ) @@ -103,7 +105,7 @@ func (d *Daemon) runGC(ctx context.Context) { // Prune stale worktree references from all bare repo caches, then evict the // caches nothing needs anymore. These live outside any workspace directory // and are never reclaimed by the task walk above. - d.pruneRepoWorktrees(root, stats) + d.pruneRepoWorktreesContext(ctx, root, stats) // Reclaim per-issue Codex session stores idle past their TTL. These live // under the shared ~/.codex home (outside WorkspacesRoot) so resume survives @@ -749,8 +751,16 @@ func (d *Daemon) cleanTaskArtifactsMatching(taskDir string, matcher artifactMatc // Non-fatal: errors during the walk are ignored so callers can report a // best-effort byte count without aborting the whole GC cycle. func dirSize(root string) int64 { + total, _ := dirSizeContext(context.Background(), root) + return total +} + +func dirSizeContext(ctx context.Context, root string) (int64, error) { var total int64 - _ = filepath.WalkDir(root, func(_ string, entry os.DirEntry, err error) error { + err := filepath.WalkDir(root, func(_ string, entry os.DirEntry, err error) error { + if ctx.Err() != nil { + return context.Cause(ctx) + } if err != nil { return nil } @@ -772,17 +782,22 @@ func dirSize(root string) int64 { } return nil }) - return total + return total, err } const ( gitCmdTimeout = 30 * time.Second gitMaintenanceTimeout = 10 * time.Minute + repoMaintenanceMarker = ".multica-maintenance-pending" ) // pruneRepoWorktrees runs `git worktree prune` on all bare repos in the cache, // then evicts the ones nothing needs anymore. func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) { + d.pruneRepoWorktreesContext(context.Background(), workspacesRoot, stats) +} + +func (d *Daemon) pruneRepoWorktreesContext(ctx context.Context, workspacesRoot string, stats *gcStats) { reposRoot := filepath.Join(workspacesRoot, reposDirName) wsEntries, err := os.ReadDir(reposRoot) if err != nil { @@ -790,6 +805,9 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) { } for _, wsEntry := range wsEntries { + if ctx.Err() != nil { + return + } if !wsEntry.IsDir() { continue } @@ -799,6 +817,9 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) { continue } for _, repoEntry := range repoEntries { + if ctx.Err() != nil { + return + } if !repoEntry.IsDir() { continue } @@ -806,7 +827,7 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) { if !isBareRepo(barePath) { continue } - d.maintainRepoCache(barePath, stats) + d.maintainRepoCache(ctx, barePath, stats) } // Drop the per-workspace directory once its last repo is gone. if remaining, err := os.ReadDir(wsRepoDir); err == nil && len(remaining) == 0 { @@ -815,17 +836,45 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) { } } -func (d *Daemon) maintainRepoCache(barePath string, stats *gcStats) { - d.withRepoLock(barePath, func() { - d.pruneWorktreeLocked(barePath) - d.evictRepoCacheLocked(barePath, stats) +func (d *Daemon) maintainRepoCache(ctx context.Context, barePath string, stats *gcStats) { + d.withRepoMaintenance(ctx, barePath, func(maintenanceCtx context.Context) { + d.pruneWorktreeLocked(maintenanceCtx, barePath) + if maintenanceCtx.Err() == nil { + d.evictRepoCacheLocked(maintenanceCtx, barePath, stats) + } }) } // pruneWorktree runs only the maintenance half — prune stale worktrees and // agent branches — without considering eviction. func (d *Daemon) pruneWorktree(barePath string) { - d.withRepoLock(barePath, func() { d.pruneWorktreeLocked(barePath) }) + d.withRepoMaintenance(context.Background(), barePath, func(ctx context.Context) { + d.pruneWorktreeLocked(ctx, barePath) + }) +} + +type repoMaintenanceBackend interface { + WithRepoMaintenance(context.Context, string, func(context.Context) error) (bool, error) +} + +// withRepoMaintenance uses the cache's foreground-priority gate when +// available. The fallback preserves test/degraded backends that predate the +// optional interface without changing the repoCacheBackend contract. +func (d *Daemon) withRepoMaintenance(ctx context.Context, barePath string, fn func(context.Context)) { + if cache, ok := d.repoCache.(repoMaintenanceBackend); ok { + ran, err := cache.WithRepoMaintenance(ctx, barePath, func(maintenanceCtx context.Context) error { + fn(maintenanceCtx) + return nil + }) + if err != nil && ctx.Err() == nil { + d.logger.Warn("gc: repo maintenance lock failed", "repo", barePath, "error", err) + } + if !ran { + d.logger.Debug("gc: repo maintenance skipped for foreground work", "repo", barePath) + } + return + } + d.withRepoLock(barePath, func() { fn(ctx) }) } // withRepoLock serializes a mutation against Sync / CreateWorktree on the same @@ -872,7 +921,7 @@ func (d *Daemon) withRepoLock(barePath string, fn func()) { // Evicting wrongly costs time, not correctness: the next task that needs the // repo takes the cache-miss path in ensureRepoReady, which re-syncs and // re-clones on demand. -func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) { +func (d *Daemon) evictRepoCacheLocked(ctx context.Context, barePath string, stats *gcStats) { if d.cfg.GCRepoTTL <= 0 { return } @@ -882,8 +931,11 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) { return } - worktrees, err := linkedWorktreeCount(barePath) + worktrees, err := linkedWorktreeCountContext(ctx, barePath) if err != nil { + if ctx.Err() != nil { + return + } d.logger.Warn("gc: worktree count failed", "repo", barePath, "error", err) return } @@ -908,7 +960,10 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) { // the repo, which on a multi-GiB cache takes long enough for a workspace to // re-attach underneath us — putting it between the check and the delete // would reopen most of the window this check exists to close. - bytes := dirSize(barePath) + bytes, err := dirSizeContext(ctx, barePath) + if err != nil { + return + } // Ask again immediately before deleting. The checks above run git and walk // the filesystem, and a workspace can re-attach this repo while they do; @@ -938,7 +993,11 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) { // worktree and marks the bare repo's own block with a `bare` line; only the // linked blocks represent checkouts that would break if the repo went away. func linkedWorktreeCount(barePath string) (int, error) { - out, err := runGitGCCommand(barePath, "worktree", "list", "--porcelain") + return linkedWorktreeCountContext(context.Background(), barePath) +} + +func linkedWorktreeCountContext(ctx context.Context, barePath string) (int, error) { + out, err := runGitGCCommandContext(ctx, barePath, "worktree", "list", "--porcelain") if err != nil { return 0, err } @@ -969,8 +1028,11 @@ func linkedWorktreeCount(barePath string) (int, error) { return count, nil } -func (d *Daemon) pruneWorktreeLocked(barePath string) { - if out, err := runGitGCCommand(barePath, "worktree", "prune"); err != nil { +func (d *Daemon) pruneWorktreeLocked(ctx context.Context, barePath string) { + if out, err := runGitGCCommandContext(ctx, barePath, "worktree", "prune"); err != nil { + if ctx.Err() != nil { + return + } d.logger.Warn("gc: worktree prune failed", "repo", barePath, "output", out, @@ -978,14 +1040,20 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) { ) } - activeBranches, err := agentWorktreeBranches(barePath) + activeBranches, err := agentWorktreeBranchesContext(ctx, barePath) if err != nil { + if ctx.Err() != nil { + return + } d.logger.Warn("gc: worktree branch scan failed", "repo", barePath, "error", err) return } - agentBranches, err := listAgentBranches(barePath) + agentBranches, err := listAgentBranchesContext(ctx, barePath) if err != nil { + if ctx.Err() != nil { + return + } d.logger.Warn("gc: agent branch scan failed", "repo", barePath, "error", err) return } @@ -995,7 +1063,10 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) { if _, ok := activeBranches[branch]; ok { continue } - if out, err := runGitGCCommand(barePath, "branch", "-D", "--", branch); err != nil { + if out, err := runGitGCCommandContext(ctx, barePath, "branch", "-D", "--", branch); err != nil { + if ctx.Err() != nil { + return + } d.logger.Warn("gc: agent branch delete failed", "repo", barePath, "branch", branch, @@ -1006,10 +1077,31 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) { } deleted++ } - if deleted == 0 { + markerPath := filepath.Join(barePath, repoMaintenanceMarker) + pending := deleted > 0 + if pending { + if err := os.WriteFile(markerPath, []byte(time.Now().UTC().Format(time.RFC3339Nano)+"\n"), 0o600); err != nil { + d.logger.Warn("gc: record pending repo maintenance failed", "repo", barePath, "error", err) + } + d.logger.Info("gc: deleted stale agent branches", "repo", barePath, "count", deleted) + } else if _, err := os.Stat(markerPath); err == nil { + pending = true + } + if !pending { + return + } + if !d.cfg.GCRepoMaintenanceEnabled { + d.logger.Debug("gc: heavy repo maintenance disabled", "repo", barePath) + return + } + // Agent CLIs can mutate linked-worktree refs directly, outside the daemon's + // in-process repository gate. Do not start heavy maintenance while any task + // is active; a new task or checkout that arrives after this check preempts + // through the maintenance context below. + if d.activeTasks.Load() > 0 { + d.logger.Debug("gc: heavy repo maintenance deferred while tasks are active", "repo", barePath) return } - d.logger.Info("gc: deleted stale agent branches", "repo", barePath, "count", deleted) // Heavier maintenance only runs when we actually removed refs, so we don't // turn every GC tick into a full `git gc --prune` on every cached repo. The @@ -1022,8 +1114,24 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) { {args: []string{"reflog", "expire", "--expire=30.days", "--all"}, timeout: gitCmdTimeout}, {args: []string{"gc", "--prune=30.days"}, timeout: gitMaintenanceTimeout}, } + completed := true for _, step := range maintenance { - if out, err := runGitCommand(barePath, step.timeout, step.args...); err != nil { + if ctx.Err() != nil || d.activeTasks.Load() > 0 { + return + } + before := snapshotRepoMaintenanceLocks(barePath) + if out, err := runGitCommandContext(ctx, barePath, step.timeout, step.args...); err != nil { + completed = false + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, repocache.ErrMaintenancePreempted) { + d.cleanupNewRepoMaintenanceLocks(barePath, before) + } + if errors.Is(context.Cause(ctx), repocache.ErrMaintenancePreempted) { + d.logger.Info("gc: git maintenance preempted for foreground work", + "repo", barePath, + "command", strings.Join(step.args, " "), + ) + return + } d.logger.Warn("gc: git maintenance failed", "repo", barePath, "command", strings.Join(step.args, " "), @@ -1032,24 +1140,107 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) { ) } } + if completed { + if err := os.Remove(markerPath); err != nil && !os.IsNotExist(err) { + d.logger.Warn("gc: clear pending repo maintenance failed", "repo", barePath, "error", err) + } + } } func runGitGCCommand(barePath string, args ...string) (string, error) { - return runGitCommand(barePath, gitCmdTimeout, args...) + return runGitGCCommandContext(context.Background(), barePath, args...) } func runGitCommand(barePath string, timeout time.Duration, args ...string) (string, error) { - ctx, cancel := context.WithTimeout(context.Background(), timeout) + return runGitCommandContext(context.Background(), barePath, timeout, args...) +} + +func runGitGCCommandContext(ctx context.Context, barePath string, args ...string) (string, error) { + return runGitCommandContext(ctx, barePath, gitCmdTimeout, args...) +} + +func runGitCommandContext(parent context.Context, barePath string, timeout time.Duration, args ...string) (string, error) { + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() cmdArgs := append([]string{"-C", barePath}, args...) - cmd := exec.CommandContext(ctx, "git", cmdArgs...) - out, err := cmd.CombinedOutput() + cmd := exec.Command("git", cmdArgs...) + out, err := processtree.CombinedOutput(ctx, cmd, 5*time.Second) return strings.TrimSpace(string(out)), err } +type repoMaintenanceLockSnapshot map[string]struct{} + +// snapshotRepoMaintenanceLocks records only lock paths known to be produced by +// the maintenance commands below. Cleanup later removes a path only if it did +// not exist in this snapshot and the process tree is confirmed gone. Checkout +// waits on the same repo gate, and task dispatch waits for CancelMaintenance's +// barrier, so no agent Git work can create a competing lock before cleanup. +func snapshotRepoMaintenanceLocks(barePath string) repoMaintenanceLockSnapshot { + locks := make(repoMaintenanceLockSnapshot) + for _, path := range repoMaintenanceLockPaths(barePath) { + locks[path] = struct{}{} + } + return locks +} + +func repoMaintenanceLockPaths(barePath string) []string { + var locks []string + for _, name := range []string{"gc.pid", "packed-refs.lock"} { + path := filepath.Join(barePath, name) + if info, err := os.Lstat(path); err == nil && info.Mode().IsRegular() { + locks = append(locks, path) + } + } + for _, root := range []string{ + filepath.Join(barePath, "refs"), + filepath.Join(barePath, "logs", "refs"), + filepath.Join(barePath, "objects", "info"), + filepath.Join(barePath, "objects", "pack"), + } { + _ = filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil { + return nil + } + if entry.Type()&linkedDirModes != 0 { + if entry.IsDir() { + return filepath.SkipDir + } + return nil + } + if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".lock") { + locks = append(locks, path) + } + return nil + }) + } + return locks +} + +func (d *Daemon) cleanupNewRepoMaintenanceLocks(barePath string, before repoMaintenanceLockSnapshot) { + for _, path := range repoMaintenanceLockPaths(barePath) { + if _, existed := before[path]; existed { + continue + } + rel, err := filepath.Rel(barePath, path) + if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) || filepath.IsAbs(rel) { + d.logger.Warn("gc: refused maintenance lock cleanup outside repo", "repo", barePath, "path", path) + continue + } + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + d.logger.Warn("gc: maintenance lock cleanup failed", "repo", barePath, "lock", rel, "error", err) + continue + } + d.logger.Info("gc: removed lock left by interrupted maintenance", "repo", barePath, "lock", rel) + } +} + func agentWorktreeBranches(barePath string) (map[string]struct{}, error) { - out, err := runGitGCCommand(barePath, "worktree", "list", "--porcelain") + return agentWorktreeBranchesContext(context.Background(), barePath) +} + +func agentWorktreeBranchesContext(ctx context.Context, barePath string) (map[string]struct{}, error) { + out, err := runGitGCCommandContext(ctx, barePath, "worktree", "list", "--porcelain") if err != nil { return nil, err } @@ -1069,10 +1260,14 @@ func agentWorktreeBranches(barePath string) (map[string]struct{}, error) { } func listAgentBranches(barePath string) ([]string, error) { + return listAgentBranchesContext(context.Background(), barePath) +} + +func listAgentBranchesContext(ctx context.Context, barePath string) ([]string, error) { // Trailing slash narrows the pattern to the `agent/` namespace only. Without // it, `for-each-ref` would also return a branch literally named `agent`, // which `agentWorktreeBranches` ignores — that branch would then be deleted. - out, err := runGitGCCommand(barePath, "for-each-ref", "--format=%(refname:short)", "refs/heads/agent/") + out, err := runGitGCCommandContext(ctx, barePath, "for-each-ref", "--format=%(refname:short)", "refs/heads/agent/") if err != nil { return nil, err } diff --git a/server/internal/daemon/gc_repo_evict_test.go b/server/internal/daemon/gc_repo_evict_test.go index 76233aec0..2e341b044 100644 --- a/server/internal/daemon/gc_repo_evict_test.go +++ b/server/internal/daemon/gc_repo_evict_test.go @@ -78,6 +78,26 @@ func TestEvictRepoCache_RemovesIdleDetachedRepo(t *testing.T) { } } +func TestEvictRepoCacheRunsWhileAnotherTaskIsActive(t *testing.T) { + d := newGCTestDaemon(t, http.NewServeMux()) + d.cfg.GCRepoTTL = 24 * time.Hour + + wsID := "11111111-1111-1111-1111-111111111111" + barePath := newEvictTestRepo(t, d, wsID, testRepoURL) + writeLastUsed(t, barePath, time.Now().Add(-48*time.Hour)) + d.activeTasks.Add(1) + defer d.activeTasks.Add(-1) + + stats := runRepoGC(d) + + if _, err := os.Stat(barePath); !os.IsNotExist(err) { + t.Fatalf("expected eligible detached repo to be evicted while another task is active, stat err = %v", err) + } + if stats.repoCachesReclaimed != 1 { + t.Errorf("repo_caches_reclaimed = %d, want 1", stats.repoCachesReclaimed) + } +} + // TestEvictRepoCache_KeepsRepoStillAttachedToWorkspace is the load-bearing // guard. Sync re-clones every listed repo that is missing whenever a workspace // registers, which happens on every daemon start — so evicting a repo the diff --git a/server/internal/daemon/gc_test.go b/server/internal/daemon/gc_test.go index 785c9b34e..e9075677a 100644 --- a/server/internal/daemon/gc_test.go +++ b/server/internal/daemon/gc_test.go @@ -26,13 +26,14 @@ func newGCTestDaemon(t *testing.T, handler http.Handler) *Daemon { root := t.TempDir() cfg := Config{ - WorkspacesRoot: root, - GCEnabled: true, - GCInterval: 1 * time.Hour, - GCTTL: 5 * 24 * time.Hour, - GCOrphanTTL: 30 * 24 * time.Hour, - GCArtifactTTL: 12 * time.Hour, - GCArtifactPatterns: []string{"node_modules", ".next", ".turbo"}, + WorkspacesRoot: root, + GCEnabled: true, + GCInterval: 1 * time.Hour, + GCTTL: 5 * 24 * time.Hour, + GCOrphanTTL: 30 * 24 * time.Hour, + GCArtifactTTL: 12 * time.Hour, + GCArtifactPatterns: []string{"node_modules", ".next", ".turbo"}, + GCRepoMaintenanceEnabled: true, } d := New(cfg, slog.Default()) d.client = NewClient(srv.URL) @@ -1053,6 +1054,28 @@ func TestPruneWorktree_RemovesOnlyStaleAgentBranches(t *testing.T) { } } +func TestMaintainRepoCacheRunsLightCleanupWhileTaskActive(t *testing.T) { + t.Parallel() + + d := newGCTestDaemon(t, http.NewServeMux()) + sourceRepo := createGCGitRepo(t) + barePath := filepath.Join(t.TempDir(), "cache.git") + runGitForGC(t, "", "clone", "--bare", sourceRepo, barePath) + const staleBranch = "agent/stale/87654321" + runGitForGC(t, "", "-C", barePath, "branch", staleBranch, "HEAD") + + d.activeTasks.Add(1) + defer d.activeTasks.Add(-1) + d.maintainRepoCache(context.Background(), barePath, &gcStats{}) + + if gitRefExists(t, barePath, "refs/heads/"+staleBranch) { + t.Fatalf("expected stale branch %q to be deleted while a task is active", staleBranch) + } + if _, err := os.Stat(filepath.Join(barePath, repoMaintenanceMarker)); err != nil { + t.Fatalf("heavy maintenance should remain pending while a task is active: %v", err) + } +} + // TestPruneWorktree_IgnoresLiteralAgentBranch ensures the GC pattern is scoped // to the `agent/` namespace. A repo whose only `agent`-shaped ref is the // literal `refs/heads/agent` (no slash) must be left untouched — the @@ -1111,6 +1134,29 @@ func TestPruneWorktree_SkipsMaintenanceWhenNothingDeleted(t *testing.T) { } } +func TestPruneWorktree_RetriesPendingMaintenanceWithoutAnotherBranchDeletion(t *testing.T) { + t.Parallel() + + d := newGCTestDaemon(t, http.NewServeMux()) + sourceRepo := createGCGitRepo(t) + barePath := filepath.Join(t.TempDir(), "cache.git") + runGitForGC(t, "", "clone", "--bare", sourceRepo, barePath) + + sentinelPath := writeOldLooseBlob(t, barePath, "pending-maintenance", 60*24*time.Hour) + markerPath := filepath.Join(barePath, repoMaintenanceMarker) + if err := os.WriteFile(markerPath, []byte("pending\n"), 0o600); err != nil { + t.Fatal(err) + } + + d.pruneWorktree(barePath) + if _, err := os.Stat(sentinelPath); !os.IsNotExist(err) { + t.Fatalf("expected pending maintenance to prune sentinel, stat err=%v", err) + } + if _, err := os.Stat(markerPath); !os.IsNotExist(err) { + t.Fatalf("expected completed maintenance to clear pending marker, stat err=%v", err) + } +} + // writeOldLooseBlob writes a dangling loose-object blob to the bare repo and // backdates its mtime so `git gc --prune=30.days` will consider it prunable. // Returns the absolute path to the loose object on disk. @@ -1137,7 +1183,7 @@ func writeOldLooseBlob(t *testing.T, barePath, content string, age time.Duration return loose } -func TestPruneWorktree_SerializesWithCreateWorktree(t *testing.T) { +func TestPruneWorktree_LegacyCacheFallbackSerializesWithCreateWorktree(t *testing.T) { t.Parallel() d := newGCTestDaemon(t, http.NewServeMux()) @@ -1209,6 +1255,131 @@ func TestPruneWorktree_SerializesWithCreateWorktree(t *testing.T) { } } +func TestCleanupNewRepoMaintenanceLocksPreservesPreexistingFiles(t *testing.T) { + t.Parallel() + + d := newGCTestDaemon(t, http.NewServeMux()) + barePath := t.TempDir() + preexisting := filepath.Join(barePath, "logs", "refs", "heads", "main.lock") + if err := os.MkdirAll(filepath.Dir(preexisting), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(preexisting, []byte("existing"), 0o644); err != nil { + t.Fatal(err) + } + before := snapshotRepoMaintenanceLocks(barePath) + + created := []string{ + filepath.Join(barePath, "refs", "remotes", "origin", "main.lock"), + filepath.Join(barePath, "objects", "pack", "multi-pack-index.lock"), + filepath.Join(barePath, "packed-refs.lock"), + filepath.Join(barePath, "gc.pid"), + } + for _, path := range created { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("new"), 0o644); err != nil { + t.Fatal(err) + } + } + outsideWhitelist := filepath.Join(barePath, "hooks", "user.lock") + if err := os.MkdirAll(filepath.Dir(outsideWhitelist), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(outsideWhitelist, []byte("keep"), 0o644); err != nil { + t.Fatal(err) + } + + d.cleanupNewRepoMaintenanceLocks(barePath, before) + for _, path := range created { + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("new maintenance lock %s survived, stat error = %v", path, err) + } + } + for _, path := range []string{preexisting, outsideWhitelist} { + if _, err := os.Stat(path); err != nil { + t.Errorf("unowned lock %s was removed: %v", path, err) + } + } +} + +func TestPruneWorktreePreemptionCleansLocksBeforeTaskStarts(t *testing.T) { + if _, err := os.Stat("/bin/sh"); err != nil { + t.Skipf("requires a POSIX shell: %v", err) + } + realGit, err := exec.LookPath("git") + if err != nil { + t.Skipf("git not found: %v", err) + } + + d := newGCTestDaemon(t, http.NewServeMux()) + sourceRepo := createGCGitRepo(t) + cache := repocache.New(filepath.Join(d.cfg.WorkspacesRoot, ".repos"), slog.Default()) + if err := cache.Sync("ws1", []repocache.RepoInfo{{URL: sourceRepo}}); err != nil { + t.Fatalf("cache sync failed: %v", err) + } + d.repoCache = cache + barePath := cache.Lookup("ws1", sourceRepo) + runGitForGC(t, "", "-C", barePath, "branch", "agent/stale/87654321", "HEAD") + + lockPath := filepath.Join(barePath, "refs", "remotes", "origin", "main.lock") + if err := os.MkdirAll(filepath.Dir(lockPath), 0o755); err != nil { + t.Fatal(err) + } + startedPath := filepath.Join(t.TempDir(), "maintenance-started") + binDir := t.TempDir() + fakeGit := filepath.Join(binDir, "git") + script := `#!/bin/sh +if [ "$3" = "reflog" ] && [ "$4" = "expire" ]; then + : > "$2/refs/remotes/origin/main.lock" + : > "$MULTICA_TEST_MAINTENANCE_STARTED" + trap 'exit 143' TERM INT + while :; do sleep 1; done +fi +exec "$MULTICA_TEST_REAL_GIT" "$@" +` + if err := os.WriteFile(fakeGit, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("MULTICA_TEST_REAL_GIT", realGit) + t.Setenv("MULTICA_TEST_MAINTENANCE_STARTED", startedPath) + t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) + + pruneDone := make(chan struct{}) + go func() { + d.pruneWorktree(barePath) + close(pruneDone) + }() + deadline := time.Now().Add(5 * time.Second) + for { + if _, err := os.Stat(startedPath); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for heavy maintenance to start") + } + time.Sleep(10 * time.Millisecond) + } + + // Match production task dispatch order: mark the task active, then cancel + // maintenance and wait for its cleanup barrier before agent code starts. + d.activeTasks.Add(1) + defer d.activeTasks.Add(-1) + cache.CancelMaintenance() + select { + case <-pruneDone: + case <-time.After(5 * time.Second): + t.Fatal("pruneWorktree did not finish after preemption") + } + if _, err := os.Stat(lockPath); !os.IsNotExist(err) { + t.Fatalf("maintenance lock survived real task preemption, stat error = %v", err) + } + if err := cache.Fetch(barePath); err != nil { + t.Fatalf("fetch did not recover after preempted maintenance cleanup: %v", err) + } +} + type blockingRepoCache struct { inner *repocache.Cache entered chan struct{} diff --git a/server/internal/daemon/health.go b/server/internal/daemon/health.go index f6b32c911..779fd03ec 100644 --- a/server/internal/daemon/health.go +++ b/server/internal/daemon/health.go @@ -51,9 +51,14 @@ type HealthResponse struct { // ActiveTaskCount remains the compatibility/safety count of every claimed // handleTask lifecycle. The additive counters split actual provider // execution from local-directory parking for throughput and diagnostics. - ActiveTaskCount int64 `json:"active_task_count"` - RunningTaskCount int64 `json:"running_task_count"` - ResourceWaitTaskCount int64 `json:"resource_wait_task_count"` + ActiveTaskCount int64 `json:"active_task_count"` + RunningTaskCount int64 `json:"running_task_count"` + ResourceWaitTaskCount int64 `json:"resource_wait_task_count"` + // Repo maintenance stays a liveness-safe background activity, so health + // remains HTTP 200/running. These additive counters explain degraded repo + // checkout capacity to operators without exposing local cache paths. + RepoMaintenanceActive int `json:"repo_maintenance_active,omitempty"` + RepoCheckoutWaiters int `json:"repo_checkout_waiters,omitempty"` Agents []string `json:"agents"` // SkippedAgents maps a provider that WAS discovered on this machine to the // reason the last registration round dropped it (version undetectable, @@ -97,8 +102,18 @@ type repoCheckoutRequest struct { AgentName string `json:"agent_name"` TaskID string `json:"task_id"` CheckoutMode string `json:"checkout_mode,omitempty"` + // RetryBusy is sent by clients that understand 503 + Retry-After. Older + // clients omit it and retain their historical unbounded lock-wait behavior. + RetryBusy bool `json:"retry_busy,omitempty"` } +const ( + repoCheckoutLockWaitTimeout = 10 * time.Second + repoCheckoutRetryAfter = 2 * time.Second + repoCheckoutRetryHeader = "X-Multica-Retryable" + repoCheckoutRetryValueBusy = "repo-busy" +) + // healthHandler returns the /health HTTP handler. Extracted from serveHealth // so tests can exercise it without spinning up a listener. func (d *Daemon) healthHandler(startedAt time.Time) http.HandlerFunc { @@ -149,6 +164,11 @@ func (d *Daemon) healthHandler(startedAt time.Time) http.HandlerFunc { ReloadPendingReason: d.reloadPending(), Workspaces: wsList, } + if reporter, ok := d.repoCache.(interface{ Activity() repocache.Activity }); ok { + activity := reporter.Activity() + resp.RepoMaintenanceActive = activity.MaintenanceActive + resp.RepoCheckoutWaiters = activity.ForegroundWaiters + } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(resp) @@ -234,6 +254,10 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc { } if err := d.ensureRepoReady(r.Context(), req.WorkspaceID, req.URL); err != nil { + if r.Context().Err() != nil { + d.logger.Debug("repo checkout readiness cancelled", "url", req.URL, "error", err) + return + } statusCode := http.StatusInternalServerError if errors.Is(err, ErrRepoNotConfigured) { statusCode = http.StatusBadRequest @@ -248,7 +272,7 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc { checkoutRef = d.taskRepoDefaultRef(req.WorkspaceID, req.TaskID, req.URL) } - result, err := d.repoCache.CreateWorktree(repocache.WorktreeParams{ + params := repocache.WorktreeParams{ WorkspaceID: req.WorkspaceID, RepoURL: req.URL, WorkDir: req.WorkDir, @@ -257,8 +281,30 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc { TaskID: req.TaskID, CoAuthoredByEnabled: d.workspaceCoAuthoredByEnabled(req.WorkspaceID), IsolatedGitMetadata: req.CheckoutMode == repoCheckoutModeIsolated, - }) + } + if req.RetryBusy { + params.LockWaitTimeout = repoCheckoutLockWaitTimeout + } + var result *repocache.WorktreeResult + var err error + if cache, ok := d.repoCache.(interface { + CreateWorktreeContext(context.Context, repocache.WorktreeParams) (*repocache.WorktreeResult, error) + }); ok { + result, err = cache.CreateWorktreeContext(r.Context(), params) + } else { + result, err = d.repoCache.CreateWorktree(params) + } if err != nil { + if errors.Is(err, repocache.ErrRepoBusy) && req.RetryBusy { + w.Header().Set(repoCheckoutRetryHeader, repoCheckoutRetryValueBusy) + w.Header().Set("Retry-After", fmt.Sprintf("%.0f", repoCheckoutRetryAfter.Seconds())) + http.Error(w, "repository is busy with another operation; retry later", http.StatusServiceUnavailable) + return + } + if r.Context().Err() != nil { + d.logger.Debug("repo checkout cancelled", "url", req.URL, "error", err) + return + } d.logger.Error("repo checkout failed", "url", req.URL, "error", err) http.Error(w, err.Error(), http.StatusInternalServerError) return diff --git a/server/internal/daemon/health_test.go b/server/internal/daemon/health_test.go index 4b93187f1..e8accda3b 100644 --- a/server/internal/daemon/health_test.go +++ b/server/internal/daemon/health_test.go @@ -196,6 +196,30 @@ func TestHealthHandlerActiveTaskCountTracksCounter(t *testing.T) { assertActiveTaskCount(t, handler, 0) } +func TestHealthHandlerReportsRepoCoordinationActivity(t *testing.T) { + t.Parallel() + + cache := &activityRepoCache{ + activity: repocache.Activity{MaintenanceActive: 1, ForegroundWaiters: 3}, + } + d := &Daemon{ + cfg: Config{CLIVersion: "v1.0.0"}, + repoCache: cache, + workspaces: map[string]*workspaceState{}, + logger: slog.Default(), + } + rec := httptest.NewRecorder() + d.healthHandler(time.Now()).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/health", nil)) + + var resp HealthResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.RepoMaintenanceActive != 1 || resp.RepoCheckoutWaiters != 3 { + t.Fatalf("repo activity = maintenance:%d waiters:%d, want 1/3", resp.RepoMaintenanceActive, resp.RepoCheckoutWaiters) + } +} + func TestShutdownHandlerPostCancelsDaemonContext(t *testing.T) { t.Parallel() @@ -371,7 +395,33 @@ func TestRepoCheckoutRejectsUnknownMode(t *testing.T) { } } -func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache *recordingRepoCache) *Daemon { +func TestRepoCheckoutReturnsRetryableBusyToCapableClient(t *testing.T) { + t.Parallel() + + const workspaceID = "ws-checkout" + const repoURL = "https://github.com/org/repo.git" + cache := &busyRepoCache{recordingRepoCache: recordingRepoCache{lookupPath: "/cache/org/repo.git"}} + d := newRepoCheckoutTestDaemon(t, workspaceID, repoURL, cache) + + rec := httptest.NewRecorder() + body := strings.NewReader(`{"url":"` + repoURL + `","workspace_id":"` + workspaceID + `","workdir":"/tmp/work","task_id":"task-1","retry_busy":true}`) + d.repoCheckoutHandler().ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/repo/checkout", body)) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d: %s", rec.Code, rec.Body.String()) + } + if got := rec.Header().Get("Retry-After"); got != "2" { + t.Fatalf("Retry-After = %q, want 2", got) + } + if got := rec.Header().Get(repoCheckoutRetryHeader); got != repoCheckoutRetryValueBusy { + t.Fatalf("%s = %q, want %q", repoCheckoutRetryHeader, got, repoCheckoutRetryValueBusy) + } + if got := cache.lastCreateParams().LockWaitTimeout; got != repoCheckoutLockWaitTimeout { + t.Fatalf("lock wait timeout = %s, want %s", got, repoCheckoutLockWaitTimeout) + } +} + +func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache repoCacheBackend) *Daemon { t.Helper() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet || r.URL.Path != "/api/daemon/workspaces/"+workspaceID+"/repos" { @@ -396,6 +446,24 @@ func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache } } +type busyRepoCache struct { + recordingRepoCache +} + +type activityRepoCache struct { + recordingRepoCache + activity repocache.Activity +} + +func (c *activityRepoCache) Activity() repocache.Activity { return c.activity } + +func (c *busyRepoCache) CreateWorktreeContext(_ context.Context, params repocache.WorktreeParams) (*repocache.WorktreeResult, error) { + c.mu.Lock() + c.params = append(c.params, params) + c.mu.Unlock() + return nil, repocache.ErrRepoBusy +} + type blockingLookupRepoCache struct { path string lookupSeen chan struct{} diff --git a/server/internal/daemon/processtree/controller_unix.go b/server/internal/daemon/processtree/controller_unix.go new file mode 100644 index 000000000..b75c20470 --- /dev/null +++ b/server/internal/daemon/processtree/controller_unix.go @@ -0,0 +1,78 @@ +//go:build !windows + +package processtree + +import ( + "errors" + "fmt" + "os" + "os/exec" + "syscall" + "time" +) + +const processTreeFinishTimeout = 5 * time.Second + +type controller struct{} + +func newController(cmd *exec.Cmd) (*controller, error) { + if cmd.SysProcAttr == nil { + cmd.SysProcAttr = &syscall.SysProcAttr{} + } + cmd.SysProcAttr.Setpgid = true + return &controller{}, nil +} + +func (*controller) attach(_ *exec.Cmd) error { return nil } + +func (*controller) interrupt(cmd *exec.Cmd) error { + if cmd.Process == nil { + return os.ErrProcessDone + } + if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM); err != nil { + if errors.Is(err, syscall.ESRCH) { + return nil + } + return err + } + return nil +} + +func (*controller) stop(cmd *exec.Cmd) error { + if cmd.Process == nil { + return os.ErrProcessDone + } + if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil { + if errors.Is(err, syscall.ESRCH) { + return nil + } + return cmd.Process.Kill() + } + return nil +} + +func (c *controller) finish(cmd *exec.Cmd) error { + if cmd.Process == nil { + return nil + } + pid := cmd.Process.Pid + if err := syscall.Kill(-pid, 0); errors.Is(err, syscall.ESRCH) { + return nil + } + // A normally-exited leader can still leave a descendant holding inherited + // pipes or repository locks. Kill the remaining group before returning + // ownership of the repository to another operation. + if err := syscall.Kill(-pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) { + return err + } + deadline := time.Now().Add(processTreeFinishTimeout) + for time.Now().Before(deadline) { + if err := syscall.Kill(-pid, 0); errors.Is(err, syscall.ESRCH) { + return nil + } + time.Sleep(10 * time.Millisecond) + } + return fmt.Errorf("process group %d still active after %s", pid, processTreeFinishTimeout) +} + +func (*controller) close() {} diff --git a/server/internal/daemon/processtree/controller_windows.go b/server/internal/daemon/processtree/controller_windows.go new file mode 100644 index 000000000..99af02b53 --- /dev/null +++ b/server/internal/daemon/processtree/controller_windows.go @@ -0,0 +1,130 @@ +//go:build windows + +package processtree + +import ( + "fmt" + "os/exec" + "syscall" + "time" + "unsafe" + + "golang.org/x/sys/windows" +) + +const processTreeFinishTimeout = 5 * time.Second + +type controller struct { + job windows.Handle +} + +type jobObjectBasicAccountingInformation struct { + TotalUserTime int64 + TotalKernelTime int64 + ThisPeriodTotalUserTime int64 + ThisPeriodTotalKernelTime int64 + TotalPageFaultCount uint32 + TotalProcesses uint32 + ActiveProcesses uint32 + TotalTerminatedProcesses uint32 +} + +func newController(cmd *exec.Cmd) (*controller, error) { + job, err := windows.CreateJobObject(nil, nil) + if err != nil { + return nil, err + } + info := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{} + info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if _, err := windows.SetInformationJobObject( + job, + windows.JobObjectExtendedLimitInformation, + uintptr(unsafe.Pointer(&info)), + uint32(unsafe.Sizeof(info)), + ); err != nil { + windows.CloseHandle(job) + return nil, fmt.Errorf("set KILL_ON_JOB_CLOSE: %w", err) + } + if cmd.SysProcAttr == nil { + cmd.SysProcAttr = &syscall.SysProcAttr{} + } + cmd.SysProcAttr.HideWindow = true + return &controller{job: job}, nil +} + +func (c *controller) attach(cmd *exec.Cmd) error { + process, err := windows.OpenProcess( + windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, + false, + uint32(cmd.Process.Pid), + ) + if err != nil { + return fmt.Errorf("open process: %w", err) + } + defer windows.CloseHandle(process) + if err := windows.AssignProcessToJobObject(c.job, process); err != nil { + return fmt.Errorf("assign process to job object: %w", err) + } + return nil +} + +func (c *controller) stop(_ *exec.Cmd) error { + if err := windows.TerminateJobObject(c.job, 1); err != nil { + active, queryErr := c.activeProcesses() + if queryErr == nil && active == 0 { + return nil + } + return err + } + return nil +} + +// Windows Job Objects have no SIGTERM equivalent; terminating the job is the +// graceful-attempt and hard-stop operation alike. +func (c *controller) interrupt(cmd *exec.Cmd) error { return c.stop(cmd) } + +func (c *controller) finish(_ *exec.Cmd) error { + active, err := c.activeProcesses() + if err != nil { + return err + } + if active > 0 { + if err := windows.TerminateJobObject(c.job, 1); err != nil { + return err + } + } + deadline := time.Now().Add(processTreeFinishTimeout) + for time.Now().Before(deadline) { + active, err = c.activeProcesses() + if err != nil { + return err + } + if active == 0 { + return nil + } + time.Sleep(10 * time.Millisecond) + } + return fmt.Errorf("process job still active after %s", processTreeFinishTimeout) +} + +func (c *controller) activeProcesses() (uint32, error) { + var info jobObjectBasicAccountingInformation + if err := windows.QueryInformationJobObject( + c.job, + windows.JobObjectBasicAccountingInformation, + uintptr(unsafe.Pointer(&info)), + uint32(unsafe.Sizeof(info)), + nil, + ); err != nil { + return 0, fmt.Errorf("query job object members: %w", err) + } + return info.ActiveProcesses, nil +} + +func (c *controller) close() { + if c.job == 0 { + return + } + _ = windows.CloseHandle(c.job) + c.job = 0 +} diff --git a/server/internal/daemon/processtree/run.go b/server/internal/daemon/processtree/run.go new file mode 100644 index 000000000..35be00762 --- /dev/null +++ b/server/internal/daemon/processtree/run.go @@ -0,0 +1,92 @@ +// Package processtree runs bounded helper commands whose descendants must not +// survive cancellation. It uses a Unix process group or a Windows Job Object. +package processtree + +import ( + "bytes" + "context" + "errors" + "fmt" + "os/exec" + "time" +) + +const gracefulStopTimeout = time.Second + +// CombinedOutput runs an unstarted command and returns its combined output. +// Cancellation terminates the entire process tree, waits for it to disappear, +// and returns the context cause rather than a platform-specific exit status. +func CombinedOutput(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) ([]byte, error) { + var output bytes.Buffer + cmd.Stdout = &output + cmd.Stderr = &output + err := run(ctx, cmd, waitDelay) + return output.Bytes(), err +} + +// Output is the process-tree-safe equivalent of exec.Cmd.Output. +func Output(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) ([]byte, error) { + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + err := run(ctx, cmd, waitDelay) + if exitErr := (*exec.ExitError)(nil); errors.As(err, &exitErr) { + exitErr.Stderr = append([]byte(nil), stderr.Bytes()...) + } + return stdout.Bytes(), err +} + +// Run executes an unstarted command while owning its complete process tree. +func Run(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) error { + return run(ctx, cmd, waitDelay) +} + +func run(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) error { + if err := ctx.Err(); err != nil { + return context.Cause(ctx) + } + controller, err := newController(cmd) + if err != nil { + return fmt.Errorf("create process-tree controller: %w", err) + } + defer controller.close() + + cmd.WaitDelay = waitDelay + if err := cmd.Start(); err != nil { + return err + } + if err := controller.attach(cmd); err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = controller.finish(cmd) + return fmt.Errorf("attach process tree: %w", err) + } + + waitDone := make(chan error, 1) + go func() { waitDone <- cmd.Wait() }() + + var stopErr error + cancelled := false + select { + case err = <-waitDone: + case <-ctx.Done(): + cancelled = true + stopErr = controller.interrupt(cmd) + timer := time.NewTimer(gracefulStopTimeout) + select { + case err = <-waitDone: + timer.Stop() + case <-timer.C: + stopErr = errors.Join(stopErr, controller.stop(cmd)) + err = <-waitDone + } + } + finishErr := controller.finish(cmd) + if lifecycleErr := errors.Join(stopErr, finishErr); lifecycleErr != nil { + return fmt.Errorf("stop process tree: %w", lifecycleErr) + } + if cancelled { + return context.Cause(ctx) + } + return err +} diff --git a/server/internal/daemon/processtree/run_unix_test.go b/server/internal/daemon/processtree/run_unix_test.go new file mode 100644 index 000000000..e792fa56c --- /dev/null +++ b/server/internal/daemon/processtree/run_unix_test.go @@ -0,0 +1,41 @@ +//go:build !windows + +package processtree + +import ( + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +func TestCombinedOutputKillsDescendantsHoldingOutput(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + marker := filepath.Join(dir, "descendant-survived") + script := filepath.Join(dir, "helper.sh") + body := "#!/bin/sh\n(sleep 0.5; echo leaked > \"$1\") &\nwait\n" + if err := os.WriteFile(script, []byte(body), 0o755); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + started := time.Now() + _, err := CombinedOutput(ctx, exec.Command(script, marker), 250*time.Millisecond) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("CombinedOutput error = %v, want deadline exceeded", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("CombinedOutput took %s, want a hard bounded return", elapsed) + } + + time.Sleep(600 * time.Millisecond) + if _, err := os.Stat(marker); !os.IsNotExist(err) { + t.Fatalf("descendant survived process-tree cancellation, stat error = %v", err) + } +} diff --git a/server/internal/daemon/repocache/cache.go b/server/internal/daemon/repocache/cache.go index a8bea6d4d..5b0c8b852 100644 --- a/server/internal/daemon/repocache/cache.go +++ b/server/internal/daemon/repocache/cache.go @@ -4,6 +4,7 @@ package repocache import ( "context" + "errors" "fmt" "log/slog" "net/url" @@ -16,6 +17,8 @@ import ( "strings" "sync" "time" + + "github.com/multica-ai/multica/server/internal/daemon/processtree" ) // gitEnv returns an environment for git subprocesses that contact remotes. @@ -65,23 +68,30 @@ var agentGitExcludePatterns = []string{ const repoCacheGitTimeout = 10 * time.Minute -func newGitCommand(ctx context.Context, args ...string) *exec.Cmd { - cmd := exec.CommandContext(ctx, "git", args...) +func newGitCommand(args ...string) *exec.Cmd { + cmd := exec.Command("git", args...) cmd.Env = gitEnv() - cmd.WaitDelay = 5 * time.Second return cmd } func runGitCombinedOutput(args ...string) ([]byte, error) { - return runGitCombinedOutputWithTimeout(repoCacheGitTimeout, args...) + return runGitCombinedOutputContext(context.Background(), args...) +} + +func runGitCombinedOutputContext(ctx context.Context, args ...string) ([]byte, error) { + return runGitCombinedOutputWithTimeoutContext(ctx, repoCacheGitTimeout, args...) } func runGitCombinedOutputWithTimeout(timeout time.Duration, args ...string) ([]byte, error) { - ctx, cancel := context.WithTimeout(context.Background(), timeout) + return runGitCombinedOutputWithTimeoutContext(context.Background(), timeout, args...) +} + +func runGitCombinedOutputWithTimeoutContext(parent context.Context, timeout time.Duration, args ...string) ([]byte, error) { + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() - cmd := newGitCommand(ctx, args...) - out, err := cmd.CombinedOutput() + cmd := newGitCommand(args...) + out, err := processtree.CombinedOutput(ctx, cmd, 5*time.Second) if ctx.Err() == context.DeadlineExceeded { return out, fmt.Errorf("git command timed out after %s: %w", timeout, ctx.Err()) } @@ -89,15 +99,23 @@ func runGitCombinedOutputWithTimeout(timeout time.Duration, args ...string) ([]b } func runGitOutput(args ...string) ([]byte, error) { - return runGitOutputWithTimeout(repoCacheGitTimeout, args...) + return runGitOutputContext(context.Background(), args...) +} + +func runGitOutputContext(ctx context.Context, args ...string) ([]byte, error) { + return runGitOutputWithTimeoutContext(ctx, repoCacheGitTimeout, args...) } func runGitOutputWithTimeout(timeout time.Duration, args ...string) ([]byte, error) { - ctx, cancel := context.WithTimeout(context.Background(), timeout) + return runGitOutputWithTimeoutContext(context.Background(), timeout, args...) +} + +func runGitOutputWithTimeoutContext(parent context.Context, timeout time.Duration, args ...string) ([]byte, error) { + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() - cmd := newGitCommand(ctx, args...) - out, err := cmd.Output() + cmd := newGitCommand(args...) + out, err := processtree.Output(ctx, cmd, 5*time.Second) if ctx.Err() == context.DeadlineExceeded { return out, fmt.Errorf("git command timed out after %s: %w", timeout, ctx.Err()) } @@ -105,15 +123,23 @@ func runGitOutputWithTimeout(timeout time.Duration, args ...string) ([]byte, err } func runGit(args ...string) error { - return runGitWithTimeout(repoCacheGitTimeout, args...) + return runGitContext(context.Background(), args...) +} + +func runGitContext(ctx context.Context, args ...string) error { + return runGitWithTimeoutContext(ctx, repoCacheGitTimeout, args...) } func runGitWithTimeout(timeout time.Duration, args ...string) error { - ctx, cancel := context.WithTimeout(context.Background(), timeout) + return runGitWithTimeoutContext(context.Background(), timeout, args...) +} + +func runGitWithTimeoutContext(parent context.Context, timeout time.Duration, args ...string) error { + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() - cmd := newGitCommand(ctx, args...) - err := cmd.Run() + cmd := newGitCommand(args...) + err := processtree.Run(ctx, cmd, 5*time.Second) if ctx.Err() == context.DeadlineExceeded { return fmt.Errorf("git command timed out after %s: %w", timeout, ctx.Err()) } @@ -140,7 +166,130 @@ type Cache struct { // hold its lock — git's own lockfiles (packed-refs.lock, config.lock, // worktree admin dirs) don't tolerate parallel mutations on the same // repo. Separate repos are independent and run concurrently. - repoLocks sync.Map // barePath -> *sync.Mutex + repoLocks sync.Map // barePath -> *repoLock +} + +// ErrRepoBusy means a foreground checkout could not acquire its repository +// within the caller's bounded wait. Callers that advertised retry support can +// turn this into a retryable HTTP response instead of waiting until their +// transport deadline expires. +var ErrRepoBusy = errors.New("repository is busy") + +// Activity is the path-free repository coordination state exposed through the +// daemon health endpoint. It is diagnostic only. +type Activity struct { + MaintenanceActive int + ForegroundWaiters int +} + +// repoLock is a foreground-priority mutex. Ordinary cache mutations serialize +// exactly as they did with sync.Mutex. Low-priority maintenance is different: +// it only starts on an idle repository and receives a context that is cancelled +// as soon as a foreground operation queues. The maintenance holder remains +// responsible for stopping its Git process tree before unlocking. +type repoLock struct { + mu sync.Mutex + held bool + maintenance bool + maintenanceCancel context.CancelCauseFunc + foregroundWaiters int + changed chan struct{} +} + +// ErrMaintenancePreempted is the cancellation cause delivered to a running +// maintenance callback when foreground work needs the same repository. +var ErrMaintenancePreempted = errors.New("repository maintenance preempted by foreground work") + +func newRepoLock() *repoLock { + return &repoLock{changed: make(chan struct{})} +} + +func (l *repoLock) LockContext(ctx context.Context) error { + l.mu.Lock() + l.foregroundWaiters++ + for { + if err := ctx.Err(); err != nil { + l.foregroundWaiters-- + l.signalLocked() + l.mu.Unlock() + return context.Cause(ctx) + } + if l.maintenanceCancel != nil { + l.maintenanceCancel(ErrMaintenancePreempted) + } + if !l.held { + l.held = true + l.maintenance = false + l.foregroundWaiters-- + l.mu.Unlock() + return nil + } + changed := l.changed + l.mu.Unlock() + select { + case <-ctx.Done(): + case <-changed: + } + l.mu.Lock() + } +} + +func (l *repoLock) Lock() { + _ = l.LockContext(context.Background()) +} + +func (l *repoLock) Unlock() { + l.mu.Lock() + if !l.held { + l.mu.Unlock() + panic("repocache: unlock of unlocked repository") + } + if l.maintenanceCancel != nil { + l.maintenanceCancel(nil) + l.maintenanceCancel = nil + } + l.held = false + l.maintenance = false + l.signalLocked() + l.mu.Unlock() +} + +func (l *repoLock) tryLockMaintenance(parent context.Context) (context.Context, bool) { + l.mu.Lock() + defer l.mu.Unlock() + if parent.Err() != nil || l.held || l.foregroundWaiters > 0 { + return nil, false + } + ctx, cancel := context.WithCancelCause(parent) + l.held = true + l.maintenance = true + l.maintenanceCancel = cancel + return ctx, true +} + +func (l *repoLock) signalLocked() { + close(l.changed) + l.changed = make(chan struct{}) +} + +func (l *repoLock) activity() (maintenance bool, waiters int) { + l.mu.Lock() + defer l.mu.Unlock() + return l.held && l.maintenance, l.foregroundWaiters +} + +func (l *repoLock) cancelMaintenanceAndWait() { + l.mu.Lock() + for l.maintenance { + if l.maintenanceCancel != nil { + l.maintenanceCancel(ErrMaintenancePreempted) + } + changed := l.changed + l.mu.Unlock() + <-changed + l.mu.Lock() + } + l.mu.Unlock() } // New creates a new repo cache rooted at the given directory. @@ -150,13 +299,41 @@ func New(root string, logger *slog.Logger) *Cache { // lockForRepo returns the mutex dedicated to the given bare repo path. See // the Cache.repoLocks field comment for semantics. -func (c *Cache) lockForRepo(barePath string) *sync.Mutex { +func (c *Cache) lockForRepo(barePath string) *repoLock { if l, ok := c.repoLocks.Load(barePath); ok { - return l.(*sync.Mutex) + return l.(*repoLock) } - newLock := &sync.Mutex{} + newLock := newRepoLock() actual, _ := c.repoLocks.LoadOrStore(barePath, newLock) - return actual.(*sync.Mutex) + return actual.(*repoLock) +} + +// Activity reports aggregate repository coordination without exposing cache +// paths. It is safe to call while operations are starting or finishing. +func (c *Cache) Activity() Activity { + var activity Activity + c.repoLocks.Range(func(_, value any) bool { + maintenance, waiters := value.(*repoLock).activity() + if maintenance { + activity.MaintenanceActive++ + } + activity.ForegroundWaiters += waiters + return true + }) + return activity +} + +// CancelMaintenance stops every active low-priority repository operation and +// waits for it to release repository ownership. Task dispatch uses this as a +// barrier before launching an agent because a task may reuse an existing +// worktree and never pass through CreateWorktree's gate. Waiting also ensures +// interrupted-maintenance lock cleanup completes before direct agent Git work +// can create its own lock files. +func (c *Cache) CancelMaintenance() { + c.repoLocks.Range(func(_, value any) bool { + value.(*repoLock).cancelMaintenanceAndWait() + return true + }) } // Sync ensures all repos for a workspace are cloned (or fetched if already cached). @@ -168,6 +345,12 @@ func (c *Cache) lockForRepo(barePath string) *sync.Mutex { // but concurrent Sync calls (different workspaces, or the same workspace // re-synced while checkouts are running) do not block each other. func (c *Cache) Sync(workspaceID string, repos []RepoInfo) error { + return c.SyncContext(context.Background(), workspaceID, repos) +} + +// SyncContext is Sync with cancellation propagated through repo lock waits, +// clone, fetch, and ref-layout migration. +func (c *Cache) SyncContext(ctx context.Context, workspaceID string, repos []RepoInfo) error { wsDir := filepath.Join(c.root, workspaceID) if err := os.MkdirAll(wsDir, 0o755); err != nil { return fmt.Errorf("create workspace cache dir: %w", err) @@ -175,17 +358,22 @@ func (c *Cache) Sync(workspaceID string, repos []RepoInfo) error { var firstErr error for _, repo := range repos { + if err := ctx.Err(); err != nil { + return context.Cause(ctx) + } if repo.URL == "" { continue } barePath := filepath.Join(wsDir, bareDirName(repo.URL)) repoLock := c.lockForRepo(barePath) - repoLock.Lock() + if err := repoLock.LockContext(ctx); err != nil { + return err + } if isBareRepo(barePath) { // Already cached — fetch latest. c.logger.Info("repo cache: fetching", "url", repo.URL, "path", barePath) - if err := gitFetch(barePath); err != nil { + if err := gitFetchContext(ctx, barePath); err != nil { c.logger.Warn("repo cache: fetch failed", "url", repo.URL, "error", err) if firstErr == nil { firstErr = err @@ -194,7 +382,7 @@ func (c *Cache) Sync(workspaceID string, repos []RepoInfo) error { } else { // Not cached — bare clone. c.logger.Info("repo cache: cloning", "url", repo.URL, "path", barePath) - if err := gitCloneBare(repo.URL, barePath); err != nil { + if err := gitCloneBareContext(ctx, repo.URL, barePath); err != nil { c.logger.Error("repo cache: clone failed", "url", repo.URL, "error", err) if firstErr == nil { firstErr = err @@ -273,12 +461,33 @@ func LastUsed(barePath string) (time.Time, bool) { // other same-repo operations that use the cache's lock (Sync, Fetch, // CreateWorktree, and daemon GC maintenance). func (c *Cache) WithRepoLock(barePath string, fn func() error) error { + return c.WithRepoLockContext(context.Background(), barePath, fn) +} + +// WithRepoLockContext is the cancellable form of WithRepoLock. +func (c *Cache) WithRepoLockContext(ctx context.Context, barePath string, fn func() error) error { repoLock := c.lockForRepo(barePath) - repoLock.Lock() + if err := repoLock.LockContext(ctx); err != nil { + return err + } defer repoLock.Unlock() return fn() } +// WithRepoMaintenance runs fn only when the repository is idle. A foreground +// waiter cancels the context passed to fn, then waits for fn to stop its process +// tree and release the repository. ran=false means maintenance was skipped +// because foreground work already owned or was waiting for the repository. +func (c *Cache) WithRepoMaintenance(ctx context.Context, barePath string, fn func(context.Context) error) (ran bool, err error) { + repoLock := c.lockForRepo(barePath) + maintenanceCtx, ok := repoLock.tryLockMaintenance(ctx) + if !ok { + return false, nil + } + defer repoLock.Unlock() + return true, fn(maintenanceCtx) +} + // Fetch runs `git fetch origin` on a cached bare clone to get latest refs. func (c *Cache) Fetch(barePath string) error { return c.WithRepoLock(barePath, func() error { @@ -374,7 +583,11 @@ func isBareRepo(path string) bool { const modernFetchRefspec = "+refs/heads/*:refs/remotes/origin/*" func gitCloneBare(url, dest string) error { - if out, err := runGitCombinedOutput("clone", "--bare", url, dest); err != nil { + return gitCloneBareContext(context.Background(), url, dest) +} + +func gitCloneBareContext(ctx context.Context, url, dest string) error { + if out, err := runGitCombinedOutputContext(ctx, "clone", "--bare", url, dest); err != nil { // Clean up partial clone. os.RemoveAll(dest) return fmt.Errorf("git clone --bare: %s: %w", strings.TrimSpace(string(out)), err) @@ -383,7 +596,7 @@ func gitCloneBare(url, dest string) error { // a mirror-style fetch refspec. Convert the bare repo to the standard // remote-tracking layout immediately so subsequent fetches write to // refs/remotes/origin/* and can't conflict with worktree-locked heads. - if err := ensureRemoteTrackingLayout(dest); err != nil { + if err := ensureRemoteTrackingLayoutContext(ctx, dest); err != nil { os.RemoveAll(dest) return fmt.Errorf("configure fetch refspec: %w", err) } @@ -400,10 +613,14 @@ func gitCloneBare(url, dest string) error { // would keep basing new worktrees on the original default branch forever // after the remote flipped. func gitFetch(barePath string) error { - if err := ensureRemoteTrackingLayout(barePath); err != nil { + return gitFetchContext(context.Background(), barePath) +} + +func gitFetchContext(ctx context.Context, barePath string) error { + if err := ensureRemoteTrackingLayoutContext(ctx, barePath); err != nil { return fmt.Errorf("ensure refspec: %w", err) } - if err := runGitFetch(barePath); err != nil { + if err := runGitFetchContext(ctx, barePath); err != nil { return err } // Refresh refs/remotes/origin/HEAD after every successful fetch. @@ -412,14 +629,18 @@ func gitFetch(barePath string) error { // getRemoteDefaultBranch, but the modern-cache default-branch-change // path (the only path that can't be recovered any other way) relies // on this call. - _ = runGit("-C", barePath, "remote", "set-head", "origin", "--auto") + _ = runGitContext(ctx, "-C", barePath, "remote", "set-head", "origin", "--auto") return nil } // runGitFetch is the raw `git fetch origin` wrapper. Callers should go through // gitFetch, which migrates legacy caches first. func runGitFetch(barePath string) error { - if out, err := runGitCombinedOutput("-C", barePath, "fetch", "origin"); err != nil { + return runGitFetchContext(context.Background(), barePath) +} + +func runGitFetchContext(ctx context.Context, barePath string) error { + if out, err := runGitCombinedOutputContext(ctx, "-C", barePath, "fetch", "origin"); err != nil { return fmt.Errorf("git fetch: %s: %w", strings.TrimSpace(string(out)), err) } return nil @@ -433,25 +654,29 @@ func runGitFetch(barePath string) error { // and runs `git remote set-head origin --auto` so getRemoteDefaultBranch can // resolve the remote's default branch. func ensureRemoteTrackingLayout(barePath string) error { - cur, err := readFetchRefspec(barePath) + return ensureRemoteTrackingLayoutContext(context.Background(), barePath) +} + +func ensureRemoteTrackingLayoutContext(ctx context.Context, barePath string) error { + cur, err := readFetchRefspecContext(ctx, barePath) if err != nil { return err } if cur == modernFetchRefspec || cur == strings.TrimPrefix(modernFetchRefspec, "+") { return nil // already modern } - if err := setFetchRefspec(barePath, modernFetchRefspec); err != nil { + if err := setFetchRefspecContext(ctx, barePath, modernFetchRefspec); err != nil { return err } // Backfill refs/remotes/origin/* by fetching with the new refspec. This // writes to the origin/* namespace, so even worktree-locked refs/heads/* // branches can't collide. - if err := runGitFetch(barePath); err != nil { + if err := runGitFetchContext(ctx, barePath); err != nil { return fmt.Errorf("backfill fetch after refspec migration: %w", err) } // Set refs/remotes/origin/HEAD so getRemoteDefaultBranch can read it. // Non-fatal: if this fails we fall back to origin/main, origin/master. - _ = runGit("-C", barePath, "remote", "set-head", "origin", "--auto") + _ = runGitContext(ctx, "-C", barePath, "remote", "set-head", "origin", "--auto") return nil } @@ -459,7 +684,11 @@ func ensureRemoteTrackingLayout(barePath string) error { // the empty string if it's not set. Distinguishes "missing" (exit 1) from // real git errors. func readFetchRefspec(barePath string) (string, error) { - out, err := runGitOutput("-C", barePath, "config", "--get", "remote.origin.fetch") + return readFetchRefspecContext(context.Background(), barePath) +} + +func readFetchRefspecContext(ctx context.Context, barePath string) (string, error) { + out, err := runGitOutputContext(ctx, "-C", barePath, "config", "--get", "remote.origin.fetch") if err != nil { if ee, ok := err.(*exec.ExitError); ok && ee.ExitCode() == 1 { return "", nil // key missing, not an error @@ -470,7 +699,11 @@ func readFetchRefspec(barePath string) (string, error) { } func setFetchRefspec(barePath, refspec string) error { - out, err := runGitCombinedOutput("-C", barePath, "config", "remote.origin.fetch", refspec) + return setFetchRefspecContext(context.Background(), barePath, refspec) +} + +func setFetchRefspecContext(ctx context.Context, barePath, refspec string) error { + out, err := runGitCombinedOutputContext(ctx, "-C", barePath, "config", "remote.origin.fetch", refspec) if err != nil { return fmt.Errorf("set remote.origin.fetch: %s: %w", strings.TrimSpace(string(out)), err) } @@ -486,6 +719,10 @@ type WorktreeParams struct { AgentName string // for branch naming TaskID string // for branch naming uniqueness CoAuthoredByEnabled bool // install prepare-commit-msg hook for Co-authored-by trailer + // LockWaitTimeout bounds only the wait for another same-repository + // operation. Zero preserves the historical unbounded wait for internal and + // older callers; retry-aware HTTP checkout requests set a finite value. + LockWaitTimeout time.Duration // IsolatedGitMetadata creates a local clone whose .git directory lives // inside WorkDir instead of a linked worktree whose gitdir lives under the // shared cache. Codex tasks need this because workspace-write keeps a @@ -506,6 +743,14 @@ type WorktreeResult struct { // at the target path (reused environment), it updates the existing worktree to // the latest remote default branch instead of failing. func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { + return c.CreateWorktreeContext(context.Background(), params) +} + +// CreateWorktreeContext is CreateWorktree with cancellation propagated through +// lock acquisition and every Git subprocess. Git work begins only after the +// lock is held, so a client that times out behind maintenance cannot leave a +// late, unwanted checkout. +func (c *Cache) CreateWorktreeContext(ctx context.Context, params WorktreeParams) (*WorktreeResult, error) { barePath := c.Lookup(params.WorkspaceID, params.RepoURL) if barePath == "" { return nil, fmt.Errorf("repo not found in cache: %s (workspace: %s)", params.RepoURL, params.WorkspaceID) @@ -515,8 +760,23 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // own lockfiles (packed-refs.lock, config.lock, worktree admin dirs) // can't tolerate parallel fetch + worktree mutations on the same repo. repoLock := c.lockForRepo(barePath) - repoLock.Lock() + lockCtx := ctx + cancel := func() {} + if params.LockWaitTimeout > 0 { + lockCtx, cancel = context.WithTimeout(ctx, params.LockWaitTimeout) + } + err := repoLock.LockContext(lockCtx) + cancel() + if err != nil { + if ctx.Err() == nil && errors.Is(err, context.DeadlineExceeded) { + return nil, fmt.Errorf("%w: %s", ErrRepoBusy, params.RepoURL) + } + return nil, err + } defer repoLock.Unlock() + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } // Stamp before doing the work, not after: a task asking for a worktree is // what "this cache is still wanted" means, whether or not the checkout @@ -528,7 +788,10 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // to the modern remote-tracking layout on first run, so subsequent fetches // never collide with the refs/heads/agent/* branches that worktree creation // locks in this same bare repo. - if err := gitFetch(barePath); err != nil { + if err := gitFetchContext(ctx, barePath); err != nil { + if ctx.Err() != nil { + return nil, context.Cause(ctx) + } // Non-fatal: preserve cached state and continue, but make the warning // loud enough that it's findable in the daemon log. The agent will // receive an older snapshot than the remote head. @@ -537,6 +800,9 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { "error", err, ) } + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } // Determine the ref to base the worktree on. By default this is the remote's // default branch (resolved internally via getRemoteDefaultBranch, which walks @@ -545,10 +811,13 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // Callers may request a specific branch, tag, or commit so review/QA agents // can inspect the exact revision without trying to mutate the daemon-owned // worktree metadata themselves. - baseRef, err := resolveBaseRef(barePath, params.Ref) + baseRef, err := resolveBaseRefContext(ctx, barePath, params.Ref) if err != nil { return nil, err } + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } // Empty here means params.Ref was unset and getRemoteDefaultBranch couldn't // resolve a default — the cache is in a state we refuse to guess from (no @@ -571,8 +840,9 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // shape even if a later task comes from an older CLI or a different runtime // that omits the mode hint. This also makes provider transitions on a reused // workdir backward compatible. - if params.IsolatedGitMetadata || isIsolatedCheckout(worktreePath) { - actualBranch, err := c.createOrUpdateIsolatedCheckout( + if params.IsolatedGitMetadata || isIsolatedCheckoutContext(ctx, worktreePath) { + actualBranch, err := c.createOrUpdateIsolatedCheckoutContext( + ctx, barePath, params.RepoURL, worktreePath, @@ -584,17 +854,20 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { } for _, pattern := range agentGitExcludePatterns { - _ = excludeFromGit(worktreePath, pattern) + _ = excludeFromGitContext(ctx, worktreePath, pattern) } if params.CoAuthoredByEnabled { - if err := installCoAuthoredByHook(worktreePath); err != nil { + if err := installCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: install co-authored-by hook failed (non-fatal)", "error", err) } } else { - if err := removeCoAuthoredByHook(worktreePath); err != nil { + if err := removeCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: remove co-authored-by hook failed (non-fatal)", "error", err) } } + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } c.logger.Info("repo checkout: isolated checkout ready", "url", params.RepoURL, @@ -608,13 +881,13 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // If worktree already exists (reused environment from a prior task), // update it to the latest remote code instead of creating a new one. if isGitWorktree(worktreePath) { - actualBranch, err := updateExistingWorktree(worktreePath, branchName, baseRef) + actualBranch, err := updateExistingWorktreeContext(ctx, worktreePath, branchName, baseRef) if err != nil { return nil, fmt.Errorf("update existing worktree: %w", err) } for _, pattern := range agentGitExcludePatterns { - _ = excludeFromGit(worktreePath, pattern) + _ = excludeFromGitContext(ctx, worktreePath, pattern) } // Install or remove the Co-authored-by hook based on the workspace @@ -623,14 +896,17 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // installed hook keeps appending the trailer to every commit even // after the user toggles the setting off. if params.CoAuthoredByEnabled { - if err := installCoAuthoredByHook(worktreePath); err != nil { + if err := installCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: install co-authored-by hook failed (non-fatal)", "error", err) } } else { - if err := removeCoAuthoredByHook(worktreePath); err != nil { + if err := removeCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: remove co-authored-by hook failed (non-fatal)", "error", err) } } + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } c.logger.Info("repo checkout: existing worktree updated", "url", params.RepoURL, @@ -647,28 +923,31 @@ func (c *Cache) CreateWorktree(params WorktreeParams) (*WorktreeResult, error) { // Create a new worktree. createWorktree may rename the branch to avoid // collisions with stale per-task refs left over from previous runs. - actualBranch, err := createWorktree(barePath, worktreePath, branchName, baseRef) + actualBranch, err := createWorktreeContext(ctx, barePath, worktreePath, branchName, baseRef) if err != nil { return nil, fmt.Errorf("create worktree: %w", err) } // Exclude agent context files from git tracking. for _, pattern := range agentGitExcludePatterns { - _ = excludeFromGit(worktreePath, pattern) + _ = excludeFromGitContext(ctx, worktreePath, pattern) } // Install or remove the Co-authored-by hook based on the workspace // setting. See the existing-worktree branch above for why removal is // required when the setting is disabled. if params.CoAuthoredByEnabled { - if err := installCoAuthoredByHook(worktreePath); err != nil { + if err := installCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: install co-authored-by hook failed (non-fatal)", "error", err) } } else { - if err := removeCoAuthoredByHook(worktreePath); err != nil { + if err := removeCoAuthoredByHookContext(ctx, worktreePath); err != nil { c.logger.Warn("repo checkout: remove co-authored-by hook failed (non-fatal)", "error", err) } } + if err := ctx.Err(); err != nil { + return nil, context.Cause(ctx) + } c.logger.Info("repo checkout: worktree created", "url", params.RepoURL, @@ -697,34 +976,38 @@ const ( // an agent's normal fetch / push commands still target GitHub rather than the // daemon-owned bare cache. func (c *Cache) createOrUpdateIsolatedCheckout(barePath, repoURL, checkoutPath, branchName, baseRef string) (string, error) { - baseCommit, err := resolveCommit(barePath, baseRef) + return c.createOrUpdateIsolatedCheckoutContext(context.Background(), barePath, repoURL, checkoutPath, branchName, baseRef) +} + +func (c *Cache) createOrUpdateIsolatedCheckoutContext(ctx context.Context, barePath, repoURL, checkoutPath, branchName, baseRef string) (string, error) { + baseCommit, err := resolveCommitContext(ctx, barePath, baseRef) if err != nil { return "", err } - if isIsolatedCheckout(checkoutPath) { - if err := setIsolatedCheckoutOrigin(checkoutPath, repoURL); err != nil { + if isIsolatedCheckoutContext(ctx, checkoutPath) { + if err := setIsolatedCheckoutOriginContext(ctx, checkoutPath, repoURL); err != nil { return "", err } // Idempotent, and required for a workdir that was first created while // the cache was still a full clone: without it, a checkout backed by a // blobless cache resolves missing blobs to nothing instead of fetching. - if isPartialClone(barePath) { - if err := configurePromisorRemote(checkoutPath); err != nil { + if isPartialCloneContext(ctx, barePath) { + if err := configurePromisorRemoteContext(ctx, checkoutPath); err != nil { return "", err } } - if err := syncIsolatedCheckoutRefs(barePath, checkoutPath, baseRef); err != nil { + if err := syncIsolatedCheckoutRefsContext(ctx, barePath, checkoutPath, baseRef); err != nil { return "", err } - actualBranch, err := updateExistingWorktree(checkoutPath, branchName, baseCommit) + actualBranch, err := updateExistingWorktreeContext(ctx, checkoutPath, branchName, baseCommit) if err != nil { return "", err } // Drop earlier tasks' agent/* heads so a reused workdir doesn't grow a // new local branch on every checkout. Non-fatal: leftover branches are // harmless clutter and must never fail the checkout. - if err := deleteStaleAgentBranches(checkoutPath, actualBranch); err != nil { + if err := deleteStaleAgentBranchesContext(ctx, checkoutPath, actualBranch); err != nil { c.logger.Warn("repo checkout: prune stale branches failed (non-fatal)", "error", err) } return actualBranch, nil @@ -733,7 +1016,7 @@ func (c *Cache) createOrUpdateIsolatedCheckout(barePath, repoURL, checkoutPath, // linked worktree. Remove it through Git (so the shared admin record is // cleaned too), then recreate the same checkout path with local metadata. if isGitWorktree(checkoutPath) { - if err := removeLinkedWorktree(barePath, checkoutPath); err != nil { + if err := removeLinkedWorktreeContext(ctx, barePath, checkoutPath); err != nil { return "", err } } @@ -743,11 +1026,15 @@ func (c *Cache) createOrUpdateIsolatedCheckout(barePath, repoURL, checkoutPath, return "", fmt.Errorf("stat checkout path: %w", err) } - return createIsolatedCheckout(barePath, repoURL, checkoutPath, branchName, baseRef, baseCommit) + return createIsolatedCheckoutContext(ctx, barePath, repoURL, checkoutPath, branchName, baseRef, baseCommit) } func removeLinkedWorktree(barePath, checkoutPath string) error { - out, err := runGitOutput("-C", checkoutPath, "rev-parse", "--git-common-dir") + return removeLinkedWorktreeContext(context.Background(), barePath, checkoutPath) +} + +func removeLinkedWorktreeContext(ctx context.Context, barePath, checkoutPath string) error { + out, err := runGitOutputContext(ctx, "-C", checkoutPath, "rev-parse", "--git-common-dir") if err != nil { return fmt.Errorf("resolve linked worktree common dir: %w", err) } @@ -758,7 +1045,7 @@ func removeLinkedWorktree(barePath, checkoutPath string) error { if !sameResolvedPath(commonDir, barePath) { return fmt.Errorf("linked worktree common dir %s does not match cache %s", commonDir, barePath) } - if out, err := runGitCombinedOutput("-C", barePath, "worktree", "remove", "--force", checkoutPath); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", barePath, "worktree", "remove", "--force", checkoutPath); err != nil { return fmt.Errorf("remove linked worktree: %s: %w", strings.TrimSpace(string(out)), err) } return nil @@ -802,7 +1089,12 @@ func localCloneArgs(goos, barePath, checkoutPath string) []string { } func createIsolatedCheckout(barePath, repoURL, checkoutPath, branchName, baseRef, baseCommit string) (_ string, retErr error) { - if out, err := runGitCombinedOutput( + return createIsolatedCheckoutContext(context.Background(), barePath, repoURL, checkoutPath, branchName, baseRef, baseCommit) +} + +func createIsolatedCheckoutContext(ctx context.Context, barePath, repoURL, checkoutPath, branchName, baseRef, baseCommit string) (_ string, retErr error) { + if out, err := runGitCombinedOutputContext( + ctx, localCloneArgs(runtime.GOOS, barePath, checkoutPath)..., ); err != nil { // Do not remove checkoutPath here. A different repository with the @@ -824,32 +1116,32 @@ func createIsolatedCheckout(barePath, repoURL, checkoutPath, branchName, baseRef // success with every file "deleted" rather than as an error. Pointing // origin at the real remote and restoring the promisor config first lets // the checkout below lazily fetch what it needs. - if out, err := runGitCombinedOutput("-C", checkoutPath, "remote", "remove", isolatedCacheRemoteName); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", checkoutPath, "remote", "remove", isolatedCacheRemoteName); err != nil { return "", fmt.Errorf("remove cache remote: %s: %w", strings.TrimSpace(string(out)), err) } - if out, err := runGitCombinedOutput("-C", checkoutPath, "remote", "add", "origin", repoURL); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", checkoutPath, "remote", "add", "origin", repoURL); err != nil { return "", fmt.Errorf("add origin remote: %s: %w", strings.TrimSpace(string(out)), err) } - if isPartialClone(barePath) { - if err := configurePromisorRemote(checkoutPath); err != nil { + if isPartialCloneContext(ctx, barePath) { + if err := configurePromisorRemoteContext(ctx, checkoutPath); err != nil { return "", err } } - if out, err := runGitCombinedOutput("-C", checkoutPath, "checkout", "--detach", baseCommit); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", checkoutPath, "checkout", "--detach", baseCommit); err != nil { return "", fmt.Errorf("git checkout --detach: %s: %w", strings.TrimSpace(string(out)), err) } - if err := deleteAllLocalBranches(checkoutPath); err != nil { + if err := deleteAllLocalBranchesContext(ctx, checkoutPath); err != nil { return "", err } - if err := syncIsolatedCheckoutRefs(barePath, checkoutPath, baseRef); err != nil { + if err := syncIsolatedCheckoutRefsContext(ctx, barePath, checkoutPath, baseRef); err != nil { return "", err } - if out, err := runGitCombinedOutput("-C", checkoutPath, "config", isolatedCheckoutConfigKey, isolatedCheckoutConfigValue); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", checkoutPath, "config", isolatedCheckoutConfigKey, isolatedCheckoutConfigValue); err != nil { return "", fmt.Errorf("mark isolated checkout: %s: %w", strings.TrimSpace(string(out)), err) } - actualBranch, err := checkoutNewBranch(checkoutPath, branchName, baseCommit) + actualBranch, err := checkoutNewBranchContext(ctx, checkoutPath, branchName, baseCommit) if err != nil { return "", err } @@ -858,7 +1150,11 @@ func createIsolatedCheckout(barePath, repoURL, checkoutPath, branchName, baseRef } func resolveCommit(repoPath, ref string) (string, error) { - out, err := runGitOutput("-C", repoPath, "rev-parse", "--verify", ref+"^{commit}") + return resolveCommitContext(context.Background(), repoPath, ref) +} + +func resolveCommitContext(ctx context.Context, repoPath, ref string) (string, error) { + out, err := runGitOutputContext(ctx, "-C", repoPath, "rev-parse", "--verify", ref+"^{commit}") if err != nil { return "", fmt.Errorf("resolve checkout base %q: %w", ref, err) } @@ -870,11 +1166,15 @@ func resolveCommit(repoPath, ref string) (string, error) { } func isIsolatedCheckout(path string) bool { + return isIsolatedCheckoutContext(context.Background(), path) +} + +func isIsolatedCheckoutContext(ctx context.Context, path string) bool { info, err := os.Stat(filepath.Join(path, ".git")) if err != nil || !info.IsDir() { return false } - out, err := runGitOutput("-C", path, "config", "--get", isolatedCheckoutConfigKey) + out, err := runGitOutputContext(ctx, "-C", path, "config", "--get", isolatedCheckoutConfigKey) return err == nil && strings.TrimSpace(string(out)) == isolatedCheckoutConfigValue } @@ -886,7 +1186,11 @@ const partialCloneFilter = "blob:none" // isPartialClone reports whether a repository was created as a partial clone, // i.e. whether git will lazily fetch missing objects from its promisor remote. func isPartialClone(repoPath string) bool { - out, err := runGitOutputWithTimeout(30*time.Second, "-C", repoPath, "config", "--get", "remote.origin.promisor") + return isPartialCloneContext(context.Background(), repoPath) +} + +func isPartialCloneContext(ctx context.Context, repoPath string) bool { + out, err := runGitOutputWithTimeoutContext(ctx, 30*time.Second, "-C", repoPath, "config", "--get", "remote.origin.promisor") if err != nil { return false } @@ -899,12 +1203,16 @@ func isPartialClone(repoPath string) bool { // `git clone --filter=blob:none` writes; `git clone --local` does not copy // them across, which is why they have to be restored by hand. func configurePromisorRemote(repoPath string) error { + return configurePromisorRemoteContext(context.Background(), repoPath) +} + +func configurePromisorRemoteContext(ctx context.Context, repoPath string) error { settings := [][2]string{ {"remote.origin.promisor", "true"}, {"remote.origin.partialclonefilter", partialCloneFilter}, } for _, kv := range settings { - if out, err := runGitCombinedOutput("-C", repoPath, "config", kv[0], kv[1]); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", repoPath, "config", kv[0], kv[1]); err != nil { return fmt.Errorf("set %s: %s: %w", kv[0], strings.TrimSpace(string(out)), err) } } @@ -912,7 +1220,11 @@ func configurePromisorRemote(repoPath string) error { } func setIsolatedCheckoutOrigin(path, repoURL string) error { - out, err := runGitCombinedOutput("-C", path, "remote", "set-url", "origin", repoURL) + return setIsolatedCheckoutOriginContext(context.Background(), path, repoURL) +} + +func setIsolatedCheckoutOriginContext(ctx context.Context, path, repoURL string) error { + out, err := runGitCombinedOutputContext(ctx, "-C", path, "remote", "set-url", "origin", repoURL) if err != nil { return fmt.Errorf("set origin remote: %s: %w", strings.TrimSpace(string(out)), err) } @@ -924,16 +1236,20 @@ func setIsolatedCheckoutOrigin(path, repoURL string) error { // a reused checkout can move to a newly-fetched commit without depending on // the cache after this function returns. func syncIsolatedCheckoutRefs(barePath, checkoutPath, baseRef string) error { + return syncIsolatedCheckoutRefsContext(context.Background(), barePath, checkoutPath, baseRef) +} + +func syncIsolatedCheckoutRefsContext(ctx context.Context, barePath, checkoutPath, baseRef string) error { refspecs := []string{ "+refs/remotes/origin/*:refs/remotes/origin/*", "+refs/tags/*:refs/tags/*", } args := []string{"-C", checkoutPath, "fetch", "--force", "--no-tags", barePath} args = append(args, refspecs...) - if out, err := runGitCombinedOutput(args...); err != nil { + if out, err := runGitCombinedOutputContext(ctx, args...); err != nil { return fmt.Errorf("sync cache refs: %s: %w", strings.TrimSpace(string(out)), err) } - if out, err := runGitCombinedOutput("-C", checkoutPath, "fetch", "--force", "--no-tags", barePath, baseRef); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", checkoutPath, "fetch", "--force", "--no-tags", barePath, baseRef); err != nil { return fmt.Errorf("fetch checkout base: %s: %w", strings.TrimSpace(string(out)), err) } return nil @@ -943,17 +1259,29 @@ func syncIsolatedCheckoutRefs(barePath, checkoutPath, baseRef string) error { // fresh local clone. The clone is detached and has no task-created branches to // preserve yet. func deleteAllLocalBranches(repoPath string) error { - return deleteLocalBranchesUnder(repoPath, "refs/heads/", "") + return deleteAllLocalBranchesContext(context.Background(), repoPath) +} + +func deleteAllLocalBranchesContext(ctx context.Context, repoPath string) error { + return deleteLocalBranchesUnderContext(ctx, repoPath, "refs/heads/", "") } // deleteStaleAgentBranches prunes branches left by earlier Multica tasks while // preserving the current task branch and every user-created local branch. func deleteStaleAgentBranches(repoPath, keepBranch string) error { - return deleteLocalBranchesUnder(repoPath, "refs/heads/agent/", "refs/heads/"+keepBranch) + return deleteStaleAgentBranchesContext(context.Background(), repoPath, keepBranch) +} + +func deleteStaleAgentBranchesContext(ctx context.Context, repoPath, keepBranch string) error { + return deleteLocalBranchesUnderContext(ctx, repoPath, "refs/heads/agent/", "refs/heads/"+keepBranch) } func deleteLocalBranchesUnder(repoPath, namespace, keepRef string) error { - out, err := runGitOutput("-C", repoPath, "for-each-ref", "--format=%(refname)", namespace) + return deleteLocalBranchesUnderContext(context.Background(), repoPath, namespace, keepRef) +} + +func deleteLocalBranchesUnderContext(ctx context.Context, repoPath, namespace, keepRef string) error { + out, err := runGitOutputContext(ctx, "-C", repoPath, "for-each-ref", "--format=%(refname)", namespace) if err != nil { return fmt.Errorf("list local branches: %w", err) } @@ -962,7 +1290,7 @@ func deleteLocalBranchesUnder(repoPath, namespace, keepRef string) error { if ref == "" || ref == keepRef { continue } - if out, err := runGitCombinedOutput("-C", repoPath, "update-ref", "-d", ref); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", repoPath, "update-ref", "-d", ref); err != nil { return fmt.Errorf("delete local branch %s: %s: %w", ref, strings.TrimSpace(string(out)), err) } } @@ -970,7 +1298,11 @@ func deleteLocalBranchesUnder(repoPath, namespace, keepRef string) error { } func checkoutNewBranch(repoPath, branchName, baseRef string) (string, error) { - out, err := runGitCombinedOutput("-C", repoPath, "checkout", "-b", branchName, baseRef) + return checkoutNewBranchContext(context.Background(), repoPath, branchName, baseRef) +} + +func checkoutNewBranchContext(ctx context.Context, repoPath, branchName, baseRef string) (string, error) { + out, err := runGitCombinedOutputContext(ctx, "-C", repoPath, "checkout", "-b", branchName, baseRef) if err == nil { return branchName, nil } @@ -979,16 +1311,20 @@ func checkoutNewBranch(repoPath, branchName, baseRef string) (string, error) { return "", wrapped } branchName = fmt.Sprintf("%s-%d", branchName, time.Now().Unix()) - if out2, err2 := runGitCombinedOutput("-C", repoPath, "checkout", "-b", branchName, baseRef); err2 != nil { + if out2, err2 := runGitCombinedOutputContext(ctx, "-C", repoPath, "checkout", "-b", branchName, baseRef); err2 != nil { return "", fmt.Errorf("git checkout -b (retry): %s: %w", strings.TrimSpace(string(out2)), err2) } return branchName, nil } func resolveBaseRef(barePath, requestedRef string) (string, error) { + return resolveBaseRefContext(context.Background(), barePath, requestedRef) +} + +func resolveBaseRefContext(ctx context.Context, barePath, requestedRef string) (string, error) { ref := strings.TrimSpace(requestedRef) if ref == "" { - return getRemoteDefaultBranch(barePath), nil + return getRemoteDefaultBranchContext(ctx, barePath), nil } // Prefer remote-tracking branches for human branch names. Then allow full @@ -999,7 +1335,7 @@ func resolveBaseRef(barePath, requestedRef string) (string, error) { ref, } for _, candidate := range candidates { - if gitRefExists(barePath, candidate+"^{commit}") { + if gitRefExistsContext(ctx, barePath, candidate+"^{commit}") { return candidate, nil } } @@ -1007,13 +1343,21 @@ func resolveBaseRef(barePath, requestedRef string) (string, error) { } func gitRefExists(repoPath, ref string) bool { - return runGit("-C", repoPath, "rev-parse", "--verify", "--quiet", ref) == nil + return gitRefExistsContext(context.Background(), repoPath, ref) +} + +func gitRefExistsContext(ctx context.Context, repoPath, ref string) bool { + return runGitContext(ctx, "-C", repoPath, "rev-parse", "--verify", "--quiet", ref) == nil } // createWorktree creates a git worktree at the given path with a new branch. // Returns the actual branch name used — which may differ from the requested // branchName if a collision was resolved by appending a timestamp suffix. func createWorktree(gitRoot, worktreePath, branchName, baseRef string) (string, error) { + return createWorktreeContext(context.Background(), gitRoot, worktreePath, branchName, baseRef) +} + +func createWorktreeContext(ctx context.Context, gitRoot, worktreePath, branchName, baseRef string) (string, error) { // Pre-check: if the worktree path already exists we would get a confusing // "already exists" error from `git worktree add` — which used to be // misclassified as a branch collision, causing the retry to leak branches @@ -1023,11 +1367,11 @@ func createWorktree(gitRoot, worktreePath, branchName, baseRef string) (string, return "", fmt.Errorf("worktree path already exists and is not a valid git worktree: %s", worktreePath) } - err := runWorktreeAdd(gitRoot, worktreePath, branchName, baseRef) + err := runWorktreeAddContext(ctx, gitRoot, worktreePath, branchName, baseRef) if err != nil && isBranchCollisionError(err) { // Branch name collision: append timestamp and retry once. branchName = fmt.Sprintf("%s-%d", branchName, time.Now().Unix()) - err = runWorktreeAdd(gitRoot, worktreePath, branchName, baseRef) + err = runWorktreeAddContext(ctx, gitRoot, worktreePath, branchName, baseRef) } if err != nil { return "", err @@ -1036,7 +1380,11 @@ func createWorktree(gitRoot, worktreePath, branchName, baseRef string) (string, } func runWorktreeAdd(gitRoot, worktreePath, branchName, baseRef string) error { - if out, err := runGitCombinedOutput("-C", gitRoot, "worktree", "add", "-b", branchName, worktreePath, baseRef); err != nil { + return runWorktreeAddContext(context.Background(), gitRoot, worktreePath, branchName, baseRef) +} + +func runWorktreeAddContext(ctx context.Context, gitRoot, worktreePath, branchName, baseRef string) error { + if out, err := runGitCombinedOutputContext(ctx, "-C", gitRoot, "worktree", "add", "-b", branchName, worktreePath, baseRef); err != nil { return fmt.Errorf("git worktree add: %s: %w", strings.TrimSpace(string(out)), err) } return nil @@ -1068,13 +1416,17 @@ func isGitWorktree(path string) bool { // the bare cache beforehand (worktrees share the same object store). // Returns the actual branch name used (may differ from input on collision). func updateExistingWorktree(worktreePath, branchName, baseRef string) (string, error) { + return updateExistingWorktreeContext(context.Background(), worktreePath, branchName, baseRef) +} + +func updateExistingWorktreeContext(ctx context.Context, worktreePath, branchName, baseRef string) (string, error) { // Discard any leftover uncommitted changes from the previous task. - if out, err := runGitCombinedOutput("-C", worktreePath, "reset", "--hard"); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", worktreePath, "reset", "--hard"); err != nil { return "", fmt.Errorf("git reset --hard: %s: %w", strings.TrimSpace(string(out)), err) } // Clean untracked files (e.g. build artifacts from previous task). - if out, err := runGitCombinedOutput("-C", worktreePath, "clean", "-fd"); err != nil { + if out, err := runGitCombinedOutputContext(ctx, "-C", worktreePath, "clean", "-fd"); err != nil { return "", fmt.Errorf("git clean -fd: %s: %w", strings.TrimSpace(string(out)), err) } @@ -1083,7 +1435,7 @@ func updateExistingWorktree(worktreePath, branchName, baseRef string) (string, e // "refs/remotes/origin/" but may be "refs/heads/" on a // legacy/migration-pending cache. Either form is valid as a checkout // startpoint. - out, err := runGitCombinedOutput("-C", worktreePath, "checkout", "-b", branchName, baseRef) + out, err := runGitCombinedOutputContext(ctx, "-C", worktreePath, "checkout", "-b", branchName, baseRef) if err == nil { return branchName, nil } @@ -1093,7 +1445,7 @@ func updateExistingWorktree(worktreePath, branchName, baseRef string) (string, e } // Branch name collision: append timestamp and retry once. branchName = fmt.Sprintf("%s-%d", branchName, time.Now().Unix()) - if out2, err2 := runGitCombinedOutput("-C", worktreePath, "checkout", "-b", branchName, baseRef); err2 != nil { + if out2, err2 := runGitCombinedOutputContext(ctx, "-C", worktreePath, "checkout", "-b", branchName, baseRef); err2 != nil { return "", fmt.Errorf("git checkout -b (retry): %s: %w", strings.TrimSpace(string(out2)), err2) } return branchName, nil @@ -1125,23 +1477,27 @@ func updateExistingWorktree(worktreePath, branchName, baseRef string) (string, e // Returns "" only when none of the above resolve — which the caller treats // as a hard error with a clear "cache has no usable refs" message. func getRemoteDefaultBranch(barePath string) string { + return getRemoteDefaultBranchContext(context.Background(), barePath) +} + +func getRemoteDefaultBranchContext(ctx context.Context, barePath string) string { // 1) Primary: refs/remotes/origin/HEAD set by `git remote set-head // origin --auto` during ensureRemoteTrackingLayout. Verify the // target actually exists — a partial set-head or a manually-broken // repo can leave a symref pointing at a deleted ref, and returning // it here would later fail in `git worktree add` with a confusing // "invalid reference" error. - if out, err := runGitOutput("-C", barePath, "symbolic-ref", "refs/remotes/origin/HEAD"); err == nil { + if out, err := runGitOutputContext(ctx, "-C", barePath, "symbolic-ref", "refs/remotes/origin/HEAD"); err == nil { ref := strings.TrimSpace(string(out)) if ref != "" { - if err := runGit("-C", barePath, "rev-parse", "--verify", ref); err == nil { + if err := runGitContext(ctx, "-C", barePath, "rev-parse", "--verify", ref); err == nil { return ref } } } // 2) Common default branch names under the origin namespace. for _, candidate := range []string{"refs/remotes/origin/main", "refs/remotes/origin/master"} { - if err := runGit("-C", barePath, "rev-parse", "--verify", candidate); err == nil { + if err := runGitContext(ctx, "-C", barePath, "rev-parse", "--verify", candidate); err == nil { return candidate } } @@ -1151,10 +1507,10 @@ func getRemoteDefaultBranch(barePath string) string { // didn't populate refs/remotes/origin/HEAD. We only return when the // matching origin/ exists, so we still pick up up-to-date code // rather than a stale local head. - bareRef := bareHeadBranch(barePath) + bareRef := bareHeadBranchContext(ctx, barePath) if bareRef != "" { originRef := "refs/remotes/origin/" + strings.TrimPrefix(bareRef, "refs/heads/") - if err := runGit("-C", barePath, "rev-parse", "--verify", originRef); err == nil { + if err := runGitContext(ctx, "-C", barePath, "rev-parse", "--verify", originRef); err == nil { return originRef } } @@ -1166,7 +1522,7 @@ func getRemoteDefaultBranch(barePath string) string { // "legacy empty" apart from "ambiguous". originCount := 0 var singleton string - if out, err := runGitOutput("-C", barePath, "for-each-ref", "--format=%(refname)", "refs/remotes/origin/"); err == nil { + if out, err := runGitOutputContext(ctx, "-C", barePath, "for-each-ref", "--format=%(refname)", "refs/remotes/origin/"); err == nil { for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { line = strings.TrimSpace(line) if line == "" || line == "refs/remotes/origin/HEAD" { @@ -1202,7 +1558,11 @@ func getRemoteDefaultBranch(barePath string) string { // modern caches should never reach this path because origin/* resolution // succeeds first. func bareHeadBranch(barePath string) string { - out, err := runGitOutput("-C", barePath, "symbolic-ref", "HEAD") + return bareHeadBranchContext(context.Background(), barePath) +} + +func bareHeadBranchContext(ctx context.Context, barePath string) string { + out, err := runGitOutputContext(ctx, "-C", barePath, "symbolic-ref", "HEAD") if err != nil { return "" } @@ -1210,7 +1570,7 @@ func bareHeadBranch(barePath string) string { if ref == "" { return "" } - if err := runGit("-C", barePath, "rev-parse", "--verify", ref); err != nil { + if err := runGitContext(ctx, "-C", barePath, "rev-parse", "--verify", ref); err != nil { return "" } return ref @@ -1266,7 +1626,11 @@ git interpret-trailers --in-place --trailer "$TRAILER" "$COMMIT_MSG_FILE" // git common directory (the bare repo for worktrees) so it applies to all // worktrees created from this cache. func installCoAuthoredByHook(worktreePath string) error { - out, err := runGitOutput("-C", worktreePath, "rev-parse", "--git-common-dir") + return installCoAuthoredByHookContext(context.Background(), worktreePath) +} + +func installCoAuthoredByHookContext(ctx context.Context, worktreePath string) error { + out, err := runGitOutputContext(ctx, "-C", worktreePath, "rev-parse", "--git-common-dir") if err != nil { return fmt.Errorf("resolve git common dir: %w", err) } @@ -1308,7 +1672,11 @@ func isDaemonInstalledHook(contents []byte) bool { // Returns nil when no hook is present or when an unrelated hook occupies // the path. func removeCoAuthoredByHook(worktreePath string) error { - out, err := runGitOutput("-C", worktreePath, "rev-parse", "--git-common-dir") + return removeCoAuthoredByHookContext(context.Background(), worktreePath) +} + +func removeCoAuthoredByHookContext(ctx context.Context, worktreePath string) error { + out, err := runGitOutputContext(ctx, "-C", worktreePath, "rev-parse", "--git-common-dir") if err != nil { return fmt.Errorf("resolve git common dir: %w", err) } @@ -1337,7 +1705,11 @@ func removeCoAuthoredByHook(worktreePath string) error { // excludeFromGit adds a pattern to the worktree's .git/info/exclude file. func excludeFromGit(worktreePath, pattern string) error { - out, err := runGitOutput("-C", worktreePath, "rev-parse", "--git-dir") + return excludeFromGitContext(context.Background(), worktreePath, pattern) +} + +func excludeFromGitContext(ctx context.Context, worktreePath, pattern string) error { + out, err := runGitOutputContext(ctx, "-C", worktreePath, "rev-parse", "--git-dir") if err != nil { return fmt.Errorf("resolve git dir: %w", err) } diff --git a/server/internal/daemon/repocache/cache_test.go b/server/internal/daemon/repocache/cache_test.go index fb4abc51b..478985f87 100644 --- a/server/internal/daemon/repocache/cache_test.go +++ b/server/internal/daemon/repocache/cache_test.go @@ -119,6 +119,203 @@ func TestRunGitOutputTimesOut(t *testing.T) { } } +func TestRepoMaintenanceYieldsToForeground(t *testing.T) { + t.Parallel() + + cache := New(t.TempDir(), testLogger()) + const barePath = "/cache/repo.git" + entered := make(chan struct{}) + maintenanceDone := make(chan error, 1) + go func() { + ran, err := cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error { + close(entered) + <-ctx.Done() + return context.Cause(ctx) + }) + if !ran && err == nil { + err = errors.New("maintenance did not run") + } + maintenanceDone <- err + }() + <-entered + + foregroundDone := make(chan error, 1) + go func() { + foregroundDone <- cache.WithRepoLockContext(context.Background(), barePath, func() error { return nil }) + }() + + select { + case err := <-foregroundDone: + if err != nil { + t.Fatalf("foreground lock: %v", err) + } + case <-time.After(time.Second): + t.Fatal("foreground work did not preempt maintenance") + } + if err := <-maintenanceDone; !errors.Is(err, ErrMaintenancePreempted) { + t.Fatalf("maintenance error = %v, want preemption", err) + } + activity := cache.Activity() + if activity.MaintenanceActive != 0 || activity.ForegroundWaiters != 0 { + t.Fatalf("activity after release = %+v, want zero", activity) + } +} + +func TestCancelMaintenanceStopsMaintenanceWithoutARepoCheckout(t *testing.T) { + t.Parallel() + + cache := New(t.TempDir(), testLogger()) + const barePath = "/cache/reused-worktree.git" + entered := make(chan struct{}) + done := make(chan error, 1) + go func() { + ran, err := cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error { + close(entered) + <-ctx.Done() + return context.Cause(ctx) + }) + if !ran && err == nil { + err = errors.New("maintenance did not run") + } + done <- err + }() + <-entered + + cache.CancelMaintenance() + select { + case err := <-done: + if !errors.Is(err, ErrMaintenancePreempted) { + t.Fatalf("maintenance error = %v, want preemption", err) + } + case <-time.After(time.Second): + t.Fatal("maintenance did not stop after task-level cancellation") + } + + if got := cache.Activity().MaintenanceActive; got != 0 { + t.Fatalf("maintenance active after cancellation = %d, want 0", got) + } +} + +func TestCancelMaintenanceWaitsForCleanupBarrier(t *testing.T) { + t.Parallel() + + cache := New(t.TempDir(), testLogger()) + const barePath = "/cache/cleanup-barrier.git" + entered := make(chan struct{}) + cancelled := make(chan struct{}) + releaseCleanup := make(chan struct{}) + maintenanceDone := make(chan struct{}) + go func() { + _, _ = cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error { + close(entered) + <-ctx.Done() + close(cancelled) + <-releaseCleanup + return context.Cause(ctx) + }) + close(maintenanceDone) + }() + <-entered + + cancelDone := make(chan struct{}) + go func() { + cache.CancelMaintenance() + close(cancelDone) + }() + <-cancelled + select { + case <-cancelDone: + t.Fatal("CancelMaintenance returned before maintenance cleanup released the repo") + case <-time.After(20 * time.Millisecond): + } + + close(releaseCleanup) + select { + case <-cancelDone: + case <-time.After(time.Second): + t.Fatal("CancelMaintenance did not return after maintenance cleanup completed") + } + <-maintenanceDone +} + +func TestCreateWorktreeContextReturnsBusyAfterBoundedLockWait(t *testing.T) { + t.Parallel() + + root := t.TempDir() + cache := New(root, testLogger()) + const ( + workspaceID = "ws-1" + repoURL = "https://github.com/org/repo.git" + ) + barePath := cache.BarePath(workspaceID, repoURL) + if err := os.MkdirAll(barePath, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(barePath, "HEAD"), []byte("ref: refs/heads/main\n"), 0o644); err != nil { + t.Fatal(err) + } + + lock := cache.lockForRepo(barePath) + lock.Lock() + defer lock.Unlock() + + _, err := cache.CreateWorktreeContext(context.Background(), WorktreeParams{ + WorkspaceID: workspaceID, + RepoURL: repoURL, + WorkDir: t.TempDir(), + LockWaitTimeout: 20 * time.Millisecond, + }) + if !errors.Is(err, ErrRepoBusy) { + t.Fatalf("CreateWorktreeContext error = %v, want ErrRepoBusy", err) + } +} + +func TestCreateWorktreeContextCancelsRunningGitProcessTree(t *testing.T) { + binDir := t.TempDir() + marker := filepath.Join(t.TempDir(), "git-descendant-survived") + script := filepath.Join(binDir, "git") + body := "#!/bin/sh\n(sleep 0.5; echo leaked > \"$MULTICA_TEST_GIT_MARKER\") &\nwait\n" + if err := os.WriteFile(script, []byte(body), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("MULTICA_TEST_GIT_MARKER", marker) + + root := t.TempDir() + cache := New(root, testLogger()) + const ( + workspaceID = "ws-1" + repoURL = "https://github.com/org/repo.git" + ) + barePath := cache.BarePath(workspaceID, repoURL) + if err := os.MkdirAll(barePath, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(barePath, "HEAD"), []byte("ref: refs/heads/main\n"), 0o644); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + started := time.Now() + _, err := cache.CreateWorktreeContext(ctx, WorktreeParams{ + WorkspaceID: workspaceID, + RepoURL: repoURL, + WorkDir: t.TempDir(), + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("CreateWorktreeContext error = %v, want deadline exceeded", err) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("CreateWorktreeContext took %s after cancellation", elapsed) + } + + time.Sleep(600 * time.Millisecond) + if _, err := os.Stat(marker); !os.IsNotExist(err) { + t.Fatalf("Git descendant survived checkout cancellation, stat error = %v", err) + } +} + func TestBareDirName(t *testing.T) { t.Parallel() tests := []struct {