From 39a9e33c375e764321e48ecb883d4133708d9cf3 Mon Sep 17 00:00:00 2001 From: J Date: Mon, 13 Jul 2026 20:09:02 +0800 Subject: [PATCH] fix(daemon): version-gate self-healed agent path; address review Addresses the PR review on MUL-4486: - Must-fix: when self-heal adopts a re-resolved binary, detect its version and enforce the same minimum-version gate registration applies, then update the detected-version cache in lockstep. Prevents (a) launching the new binary under the daemon's stale version policy (e.g. Codex sandbox policy keyed on version) and (b) silently adopting a below-minimum build that the registration path would have rejected. A candidate that can't be version-detected or is below minimum is not adopted; the stale entry is kept so it is never launched. - Coalesce concurrent per-provider heals with singleflight so a just-upgraded agent seen by many queued tasks pays for one login-shell + --version probe instead of one per task. - Reorder runTask so a custom-runtime-profile task resolves its profile path first and skips the built-in self-heal entirely (no wasted re-resolution on custom-only hosts). MUL-4486 Co-authored-by: multica-agent --- .../daemon/agent_path_selfheal_test.go | 97 +++++++++++++-- server/internal/daemon/daemon.go | 113 ++++++++++++++---- 2 files changed, 178 insertions(+), 32 deletions(-) diff --git a/server/internal/daemon/agent_path_selfheal_test.go b/server/internal/daemon/agent_path_selfheal_test.go index 44d78e302..658ec60c0 100644 --- a/server/internal/daemon/agent_path_selfheal_test.go +++ b/server/internal/daemon/agent_path_selfheal_test.go @@ -1,6 +1,8 @@ package daemon import ( + "context" + "log/slog" "os" "path/filepath" "runtime" @@ -8,6 +10,30 @@ import ( "testing" ) +// newSelfHealTestDaemon builds a Daemon with just the state resolveAgentEntry +// touches (logger + the two maps it reads/writes), so these tests don't need a +// fake HTTP server. healGroup / the mutexes are usable at their zero value. +func newSelfHealTestDaemon() *Daemon { + return &Daemon{ + logger: slog.Default(), + resolvedPaths: make(map[string]string), + agentVersions: make(map[string]string), + } +} + +// stubDetectVersionFromPath makes detectAgentVersion report the version encoded +// in an installVersionedCodex path (…/codex//bin/codex), so a heal to a +// v2 directory "detects" v2. checkAgentMinVersion is intentionally left as the +// real agent.CheckMinVersion so the min-version gate is exercised for real. +func stubDetectVersionFromPath(t *testing.T) { + t.Helper() + orig := detectAgentVersion + detectAgentVersion = func(_ context.Context, path string) (string, error) { + return filepath.Base(filepath.Dir(filepath.Dir(path))), nil + } + t.Cleanup(func() { detectAgentVersion = orig }) +} + // writeExecStub writes a runnable no-op executable at path, creating parents. func writeExecStub(t *testing.T, path string) { t.Helper() @@ -42,12 +68,14 @@ func installVersionedCodex(t *testing.T, root, version, stableBin string) string // TestResolveAgentEntry_SelfHealsAfterInPlaceUpgrade reproduces MUL-4486: a // version manager upgrades codex in place, deleting the versioned directory the // daemon pinned at startup and repointing the stable command name at the new -// version. resolveAgentEntry must re-resolve the pinned path instead of leaving -// the daemon stuck on a path that no longer exists. +// version. resolveAgentEntry must re-resolve the pinned path AND move the +// cached detected version in lockstep, so downstream policy sees the binary +// that will actually run. func TestResolveAgentEntry_SelfHealsAfterInPlaceUpgrade(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("symlink/exec-bit layout is POSIX-specific") } + stubDetectVersionFromPath(t) root := t.TempDir() stableBin := filepath.Join(root, "bin") // the stable "/opt/homebrew/bin" analogue @@ -62,12 +90,14 @@ func TestResolveAgentEntry_SelfHealsAfterInPlaceUpgrade(t *testing.T) { t.Fatalf("pinned path %q does not point into the v1 versioned dir", v1) } - d := newTestDaemon(t) + d := newSelfHealTestDaemon() + d.setAgentVersion("codex", "0.144.1") // the version detected for v1 at startup entry := AgentEntry{Path: v1, Command: "codex"} + ctx := context.Background() // While the pinned binary is present it must be returned unchanged — the // anti-PATH-redirect guarantee for the normal case. - if got := d.resolveAgentEntry("codex", entry); got.Path != v1 { + if got := d.resolveAgentEntry(ctx, "codex", entry); got.Path != v1 { t.Fatalf("live pinned path was rewritten: got %q, want %q", got.Path, v1) } @@ -80,23 +110,70 @@ func TestResolveAgentEntry_SelfHealsAfterInPlaceUpgrade(t *testing.T) { } v2 := installVersionedCodex(t, root, "0.144.3", stableBin) - got := d.resolveAgentEntry("codex", entry) + got := d.resolveAgentEntry(ctx, "codex", entry) if got.Path != v2 { t.Fatalf("self-heal resolved %q, want re-resolved v2 %q", got.Path, v2) } if !agentExecutablePresent(got.Path) { t.Fatalf("self-healed path is not runnable: %q", got.Path) } + // Must-fix (MUL-4486 review): the detected-version cache moved with the path. + if v := d.agentVersion("codex"); v != "0.144.3" { + t.Fatalf("agent version not updated in lockstep: got %q, want %q", v, "0.144.3") + } // A subsequent call with the same stale entry must reuse the remembered // heal without re-resolving from scratch. Prove it by breaking PATH: if it // re-resolved now it would miss, but the cached healed path still resolves. t.Setenv("PATH", filepath.Join(root, "empty")) - if got := d.resolveAgentEntry("codex", entry); got.Path != v2 { + if got := d.resolveAgentEntry(ctx, "codex", entry); got.Path != v2 { t.Fatalf("cached self-heal not reused: got %q, want %q", got.Path, v2) } } +// TestResolveAgentEntry_RejectsBelowMinVersionAfterUpgrade covers the review's +// second concern: if re-resolution lands on a build below the minimum supported +// version, it must NOT be adopted or launched, and the cached version must not +// be corrupted downward. +func TestResolveAgentEntry_RejectsBelowMinVersionAfterUpgrade(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink/exec-bit layout is POSIX-specific") + } + stubDetectVersionFromPath(t) // checkAgentMinVersion stays real: codex min is 0.100.0 + + root := t.TempDir() + stableBin := filepath.Join(root, "bin") + t.Setenv("PATH", stableBin) + t.Setenv("SHELL", filepath.Join(t.TempDir(), "fish")) + + v1 := installVersionedCodex(t, root, "0.144.1", stableBin) + + d := newSelfHealTestDaemon() + d.setAgentVersion("codex", "0.144.1") + entry := AgentEntry{Path: v1, Command: "codex"} + ctx := context.Background() + + // "Upgrade" actually repoints at a below-minimum build (< 0.100.0). + if err := os.RemoveAll(filepath.Join(root, "Caskroom", "codex", "0.144.1")); err != nil { + t.Fatalf("remove v1 tree: %v", err) + } + installVersionedCodex(t, root, "0.9.0", stableBin) + + got := d.resolveAgentEntry(ctx, "codex", entry) + if got.Path != v1 { + t.Fatalf("below-min build was adopted: got %q, want stale pinned %q", got.Path, v1) + } + d.resolvedPathsMu.RLock() + _, cached := d.resolvedPaths["codex"] + d.resolvedPathsMu.RUnlock() + if cached { + t.Fatalf("below-min build was cached as a healed path") + } + if v := d.agentVersion("codex"); v != "0.144.1" { + t.Fatalf("cached version was corrupted to a below-min value: got %q, want %q", v, "0.144.1") + } +} + // TestResolveAgentEntry_UninstalledLeavesEntryUnchanged verifies that when the // binary is genuinely gone (not just moved by an upgrade), resolveAgentEntry // returns the entry untouched so the downstream "executable not found" error @@ -119,9 +196,9 @@ func TestResolveAgentEntry_UninstalledLeavesEntryUnchanged(t *testing.T) { t.Fatalf("remove install root: %v", err) } - d := newTestDaemon(t) + d := newSelfHealTestDaemon() entry := AgentEntry{Path: pinned, Command: "codex"} - got := d.resolveAgentEntry("codex", entry) + got := d.resolveAgentEntry(context.Background(), "codex", entry) if got.Path != pinned { t.Fatalf("expected entry unchanged when binary is gone, got %q want %q", got.Path, pinned) } @@ -131,9 +208,9 @@ func TestResolveAgentEntry_UninstalledLeavesEntryUnchanged(t *testing.T) { // no recorded Command (e.g. a custom runtime profile carrying an absolute path) // is never re-resolved: the entry is returned as-is even when its path is gone. func TestResolveAgentEntry_NoCommandNoHeal(t *testing.T) { - d := newTestDaemon(t) + d := newSelfHealTestDaemon() entry := AgentEntry{Path: filepath.Join(t.TempDir(), "does-not-exist"), Command: ""} - if got := d.resolveAgentEntry("codex", entry); got.Path != entry.Path { + if got := d.resolveAgentEntry(context.Background(), "codex", entry); got.Path != entry.Path { t.Fatalf("entry with empty Command was rewritten: got %q, want %q", got.Path, entry.Path) } } diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index be7a2322c..af95b2c3c 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -18,6 +18,8 @@ import ( "sync/atomic" "time" + "golang.org/x/sync/singleflight" + "github.com/multica-ai/multica/server/internal/cli" "github.com/multica-ai/multica/server/internal/daemon/execenv" "github.com/multica-ai/multica/server/internal/daemon/repocache" @@ -181,6 +183,10 @@ type Daemon struct { // it without re-resolving. Keyed by provider. See MUL-4486. resolvedPathsMu sync.RWMutex resolvedPaths map[string]string + // healGroup coalesces concurrent self-heal re-resolutions per provider so a + // just-upgraded agent seen by many queued tasks at once pays for a single + // login-shell probe + version detection instead of one per task (MUL-4486). + healGroup singleflight.Group wsHBMu sync.RWMutex // guards wsHBLastAck wsHBLastAck map[string]time.Time // runtime_id -> last successful WS heartbeat ack timestamp @@ -319,16 +325,21 @@ func (d *Daemon) agentVersion(provider string) string { // Behaviour: // - Pinned Path still present -> returned unchanged. The anti-redirect // guarantee holds for the normal case: a live pinned binary is never -// second-guessed even if PATH now points elsewhere. +// second-guessed even if PATH now points elsewhere, and its cached detected +// version already corresponds to it. // - Pinned Path gone -> reuse a previously self-healed path if it is still // present, otherwise re-resolve entry.Command once (preserving the -// ~/.multica/hooks exclusion and the login-shell fallback) and remember the -// result for later calls. This reproduces exactly what a daemon restart -// would resolve, so it is no less safe than the documented restart -// workaround — only automatic. -// - Re-resolution fails (binary genuinely uninstalled) -> entry is returned -// unchanged so the downstream "executable not found" error still surfaces. -func (d *Daemon) resolveAgentEntry(provider string, entry AgentEntry) AgentEntry { +// ~/.multica/hooks exclusion and the login-shell fallback). Before adopting +// the re-resolved binary it is version-detected and run through the same +// minimum-version gate registration applies, and on success the detected +// version cache is updated in lockstep so downstream policy (e.g. the Codex +// sandbox version) matches the binary that will actually run. This +// reproduces exactly what a daemon restart would resolve, so it is no less +// safe than the documented restart workaround — only automatic. +// - Re-resolution fails, the candidate can't be version-detected, or it is +// below the minimum supported version -> entry is returned unchanged so the +// candidate is never launched and the downstream error still surfaces. +func (d *Daemon) resolveAgentEntry(ctx context.Context, provider string, entry AgentEntry) AgentEntry { if agentExecutablePresent(entry.Path) { return entry } @@ -341,22 +352,78 @@ func (d *Daemon) resolveAgentEntry(provider string, entry AgentEntry) AgentEntry return entry } - newPath, found := reresolveAgentCommand(entry.Command) - if !found { + if entry.Command == "" { return entry } + // Coalesce concurrent heals for the same provider: the first task through + // pays for the re-resolve + version detection, the rest share its result + // instead of each spawning their own login shell and `--version` probe. + command := entry.Command + v, _, _ := d.healGroup.Do(provider, func() (any, error) { + return d.healAgentPath(ctx, provider, command), nil + }) + newPath, _ := v.(string) + if newPath == "" { + return entry + } + entry.Path = newPath + return entry +} + +// healAgentPath re-resolves command for provider and, if a usable binary is +// found, records it and returns its path. "Usable" means: it resolves, its +// version can be detected, and that version meets the same minimum-version gate +// registration enforces. Adopting the path also refreshes the detected-version +// cache so callers that read d.agentVersion(provider) — notably the Codex +// sandbox policy — see the version of the binary that will actually run. It +// returns "" when nothing usable was found, so the caller keeps the (stale) +// pinned entry and the candidate is never launched. Runs under healGroup, one +// invocation at a time per provider. +func (d *Daemon) healAgentPath(ctx context.Context, provider, command string) string { + // Re-check the cache: a predecessor under the same singleflight key may have + // already populated it, or a prior heal completed between the read above and + // entering here. + d.resolvedPathsMu.RLock() + cached, ok := d.resolvedPaths[provider] + d.resolvedPathsMu.RUnlock() + if ok && agentExecutablePresent(cached) { + return cached + } + + newPath, found := reresolveAgentCommand(command) + if !found { + return "" + } + + // Verify before adopting. An in-place "upgrade" that actually repoints at an + // older or broken install must not be launched under the daemon's stale + // version policy, and must not slip past the minimum-version gate that the + // registration path applies (MUL-4486 review). + version, err := detectAgentVersion(ctx, newPath) + if err != nil { + d.logger.Warn("re-resolved agent executable failed version detection; keeping pinned path", + "provider", provider, "command", command, "new_path", newPath, "error", err) + return "" + } + if err := checkAgentMinVersion(provider, version); err != nil { + d.logger.Warn("re-resolved agent executable is below the minimum supported version; not adopting it", + "provider", provider, "command", command, "new_path", newPath, "version", version, "error", err) + return "" + } + d.resolvedPathsMu.Lock() if d.resolvedPaths == nil { d.resolvedPaths = make(map[string]string) } d.resolvedPaths[provider] = newPath d.resolvedPathsMu.Unlock() + // Keep the detected-version cache in lockstep with the path. + d.setAgentVersion(provider, version) d.logger.Info("re-resolved agent executable after pinned path vanished (in-place upgrade)", - "provider", provider, "command", entry.Command, "old_path", entry.Path, "new_path", newPath) - entry.Path = newPath - return entry + "provider", provider, "command", command, "new_path", newPath, "version", version) + return newPath } func (d *Daemon) notifyRuntimeSetChanged() { @@ -997,8 +1064,10 @@ func (d *Daemon) registerRuntimesForWorkspace(ctx context.Context, workspaceID s for name, entry := range d.cfg.Agents { // Self-heal a pinned executable path an in-place upgrade deleted // (MUL-4486) so version detection — and thus staying registered/online - // — recovers without a daemon restart. - entry = d.resolveAgentEntry(name, entry) + // — recovers without a daemon restart. resolveAgentEntry already + // version-gates the healed binary; the detect + min-version check below + // still runs to produce the version string this registration reports. + entry = d.resolveAgentEntry(ctx, name, entry) version, err := detectAgentVersion(ctx, entry.Path) if err != nil { d.logger.Warn("skip registering runtime", "name", name, "error", err) @@ -2123,7 +2192,7 @@ func (d *Daemon) handleModelList(ctx context.Context, rt Runtime, requestID stri return } // Self-heal a pinned executable path an in-place upgrade deleted (MUL-4486). - entry = d.resolveAgentEntry(rt.Provider, entry) + entry = d.resolveAgentEntry(ctx, rt.Provider, entry) models, err := agent.ListModels(ctx, rt.Provider, entry.Path) if err != nil { @@ -3544,12 +3613,6 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i defer d.clearTaskRepoRefs(task.WorkspaceID, task.ID) entry, ok := d.cfg.Agents[provider] - if ok { - // Self-heal a pinned executable path that an in-place upgrade deleted - // (MUL-4486). Done before the custom-profile override below so a - // custom runtime's own path is never second-guessed. - entry = d.resolveAgentEntry(provider, entry) - } // A custom runtime profile (MUL-3284) overrides the executable path: the // runtime's protocol_family is the provider (so agent.New still selects // the right backend), but the actual binary on PATH is the profile's @@ -3566,6 +3629,12 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i "task_id", task.ID, "runtime_id", task.RuntimeID, "provider", provider, "command_path", customSpec.path, "fixed_args", len(profileFixedArgs)) + } else if ok { + // Built-in provider: self-heal a pinned executable path that an in-place + // upgrade deleted (MUL-4486). Only reached when no custom profile owns + // the launch, so a custom runtime's path is never second-guessed and a + // custom-only host pays no wasted re-resolution. + entry = d.resolveAgentEntry(ctx, provider, entry) } if !ok { return TaskResult{}, fmt.Errorf("no agent configured for provider %q", provider)