mirror of
https://github.com/coder/coder.git
synced 2026-06-02 20:48:20 +00:00
fix(agent/agentcontext): address coder-agents-review CRF-1 through CRF-5
- CRF-1 (P2): thread quartz.Clock through PushOptions so pushWithRetry uses clock.NewTimer, making the retry test deterministic via a quartz trap instead of real sleeps. - CRF-2 (P3): remove dead skillsParentNames map; simplify isSkillsContainer to a base-name check that already covered every reachable case. - CRF-3 (P3): remove unused Snapshot.AggregateHashHex (api.go inlines hex encoding for HTTP responses). - CRF-4 (Nit): replace time.Sleep timing waits in watch_test.go and manager_test.go with Eventually-driven writes and a new Manager.Started signal channel. - CRF-5 (Nit): drop unused parameters from defaultContextRoots.
This commit is contained in:
+1
-1
@@ -459,7 +459,7 @@ func (a *agent) init() {
|
|||||||
Logger: a.logger.Named("agentcontext"),
|
Logger: a.logger.Named("agentcontext"),
|
||||||
Clock: a.clock,
|
Clock: a.clock,
|
||||||
WorkingDir: workingDirFn,
|
WorkingDir: workingDirFn,
|
||||||
BuiltinRoots: defaultContextRoots(a.contextConfig, workingDirFn),
|
BuiltinRoots: defaultContextRoots(),
|
||||||
InitialSources: initialContextSources(a.contextConfig, workingDirFn),
|
InitialSources: initialContextSources(a.contextConfig, workingDirFn),
|
||||||
AllowedRoots: defaultContextAllowedRoots(workingDirFn),
|
AllowedRoots: defaultContextAllowedRoots(workingDirFn),
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -90,10 +90,11 @@ type Manager struct {
|
|||||||
trigger chan struct{}
|
trigger chan struct{}
|
||||||
|
|
||||||
// running tracks Run lifetime.
|
// running tracks Run lifetime.
|
||||||
running bool
|
running bool
|
||||||
closed bool
|
closed bool
|
||||||
closedCh chan struct{}
|
closedCh chan struct{}
|
||||||
runDoneCh chan struct{}
|
runDoneCh chan struct{}
|
||||||
|
runStartedCh chan struct{}
|
||||||
|
|
||||||
watcher *Watcher
|
watcher *Watcher
|
||||||
}
|
}
|
||||||
@@ -135,6 +136,7 @@ func NewManager(opts ManagerOptions) (*Manager, error) {
|
|||||||
trigger: make(chan struct{}, 1),
|
trigger: make(chan struct{}, 1),
|
||||||
closedCh: make(chan struct{}),
|
closedCh: make(chan struct{}),
|
||||||
runDoneCh: make(chan struct{}),
|
runDoneCh: make(chan struct{}),
|
||||||
|
runStartedCh: make(chan struct{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, s := range opts.InitialSources {
|
for _, s := range opts.InitialSources {
|
||||||
@@ -179,6 +181,7 @@ func (m *Manager) Run(ctx context.Context) error {
|
|||||||
return xerrors.New("agentcontext: Manager already closed")
|
return xerrors.New("agentcontext: Manager already closed")
|
||||||
}
|
}
|
||||||
m.running = true
|
m.running = true
|
||||||
|
close(m.runStartedCh)
|
||||||
m.mu.Unlock()
|
m.mu.Unlock()
|
||||||
|
|
||||||
watcher, err := NewWatcher(WatcherOptions{
|
watcher, err := NewWatcher(WatcherOptions{
|
||||||
@@ -218,6 +221,14 @@ func (m *Manager) Run(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Started returns a channel that is closed once Run has
|
||||||
|
// claimed the running flag. Callers waiting to coordinate with
|
||||||
|
// the watcher loop can select on it; a closed channel never
|
||||||
|
// blocks, so this is safe to call repeatedly.
|
||||||
|
func (m *Manager) Started() <-chan struct{} {
|
||||||
|
return m.runStartedCh
|
||||||
|
}
|
||||||
|
|
||||||
// Close stops the Manager. Close is idempotent; subsequent
|
// Close stops the Manager. Close is idempotent; subsequent
|
||||||
// calls block until Run exits.
|
// calls block until Run exits.
|
||||||
func (m *Manager) Close() error {
|
func (m *Manager) Close() error {
|
||||||
|
|||||||
@@ -223,8 +223,15 @@ func TestManager_RunOnce(t *testing.T) {
|
|||||||
ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitShort))
|
ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitShort))
|
||||||
defer cancel()
|
defer cancel()
|
||||||
go func() { _ = m.Run(ctx) }()
|
go func() { _ = m.Run(ctx) }()
|
||||||
// Brief wait so Run has a chance to set running=true.
|
|
||||||
time.Sleep(50 * time.Millisecond)
|
// Wait for Run to claim the running flag, then verify the
|
||||||
|
// second call rejects with a deterministic error rather than
|
||||||
|
// racing the scheduler.
|
||||||
|
select {
|
||||||
|
case <-m.Started():
|
||||||
|
case <-ctx.Done():
|
||||||
|
t.Fatalf("manager never started: %v", ctx.Err())
|
||||||
|
}
|
||||||
|
|
||||||
err := m.Run(ctx)
|
err := m.Run(ctx)
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"golang.org/x/xerrors"
|
"golang.org/x/xerrors"
|
||||||
|
|
||||||
"cdr.dev/slog/v3"
|
"cdr.dev/slog/v3"
|
||||||
|
"github.com/coder/quartz"
|
||||||
)
|
)
|
||||||
|
|
||||||
// PushRequest is the wire-format-independent payload the
|
// PushRequest is the wire-format-independent payload the
|
||||||
@@ -58,6 +59,10 @@ type PushOptions struct {
|
|||||||
InitialBackoff time.Duration
|
InitialBackoff time.Duration
|
||||||
// MaxBackoff caps the retry wait. Default 30s.
|
// MaxBackoff caps the retry wait. Default 30s.
|
||||||
MaxBackoff time.Duration
|
MaxBackoff time.Duration
|
||||||
|
// Clock is the time source for retry backoffs. Optional;
|
||||||
|
// defaults to the Manager's clock so tests can trap waits
|
||||||
|
// with quartz instead of real sleeps.
|
||||||
|
Clock quartz.Clock
|
||||||
}
|
}
|
||||||
|
|
||||||
// RunPush ships the current snapshot to the Pusher, then ships
|
// RunPush ships the current snapshot to the Pusher, then ships
|
||||||
@@ -81,6 +86,10 @@ func (m *Manager) RunPush(ctx context.Context, p Pusher, opts PushOptions) error
|
|||||||
if maxBackoff <= 0 {
|
if maxBackoff <= 0 {
|
||||||
maxBackoff = 30 * time.Second
|
maxBackoff = 30 * time.Second
|
||||||
}
|
}
|
||||||
|
clock := opts.Clock
|
||||||
|
if clock == nil {
|
||||||
|
clock = m.clock
|
||||||
|
}
|
||||||
|
|
||||||
changes, unsub := m.SubscribeChanges()
|
changes, unsub := m.SubscribeChanges()
|
||||||
defer unsub()
|
defer unsub()
|
||||||
@@ -91,7 +100,7 @@ func (m *Manager) RunPush(ctx context.Context, p Pusher, opts PushOptions) error
|
|||||||
snap := m.Snapshot()
|
snap := m.Snapshot()
|
||||||
req := snapshotToPushRequest(snap, initial)
|
req := snapshotToPushRequest(snap, initial)
|
||||||
|
|
||||||
err := pushWithRetry(ctx, p, req, initialBackoff, maxBackoff, logger)
|
err := pushWithRetry(ctx, p, req, initialBackoff, maxBackoff, clock, logger)
|
||||||
switch {
|
switch {
|
||||||
case err == nil:
|
case err == nil:
|
||||||
initial = false
|
initial = false
|
||||||
@@ -130,6 +139,7 @@ func pushWithRetry(
|
|||||||
p Pusher,
|
p Pusher,
|
||||||
req *PushRequest,
|
req *PushRequest,
|
||||||
initialBackoff, maxBackoff time.Duration,
|
initialBackoff, maxBackoff time.Duration,
|
||||||
|
clock quartz.Clock,
|
||||||
logger slog.Logger,
|
logger slog.Logger,
|
||||||
) error {
|
) error {
|
||||||
backoff := initialBackoff
|
backoff := initialBackoff
|
||||||
@@ -155,10 +165,12 @@ func pushWithRetry(
|
|||||||
slog.F("version", req.Version),
|
slog.F("version", req.Version),
|
||||||
slog.F("backoff", backoff),
|
slog.F("backoff", backoff),
|
||||||
slog.Error(err))
|
slog.Error(err))
|
||||||
|
timer := clock.NewTimer(backoff)
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
|
timer.Stop()
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
case <-time.After(backoff):
|
case <-timer.C:
|
||||||
}
|
}
|
||||||
backoff *= 2
|
backoff *= 2
|
||||||
if backoff > maxBackoff {
|
if backoff > maxBackoff {
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
|
|
||||||
"github.com/coder/coder/v2/agent/agentcontext"
|
"github.com/coder/coder/v2/agent/agentcontext"
|
||||||
"github.com/coder/coder/v2/testutil"
|
"github.com/coder/coder/v2/testutil"
|
||||||
|
"github.com/coder/quartz"
|
||||||
)
|
)
|
||||||
|
|
||||||
// fakePusher records every push and lets the test control the
|
// fakePusher records every push and lets the test control the
|
||||||
@@ -156,6 +157,10 @@ func TestRunPush_StopsOnUnimplemented(t *testing.T) {
|
|||||||
|
|
||||||
func TestRunPush_RetriesTransientError(t *testing.T) {
|
func TestRunPush_RetriesTransientError(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
mClock := quartz.NewMock(t)
|
||||||
|
trap := mClock.Trap().NewTimer()
|
||||||
|
defer trap.Close()
|
||||||
|
|
||||||
m := newTestManager(t, agentcontext.ManagerOptions{
|
m := newTestManager(t, agentcontext.ManagerOptions{
|
||||||
WorkingDir: func() string { return t.TempDir() },
|
WorkingDir: func() string { return t.TempDir() },
|
||||||
})
|
})
|
||||||
@@ -169,11 +174,17 @@ func TestRunPush_RetriesTransientError(t *testing.T) {
|
|||||||
go func() {
|
go func() {
|
||||||
pushDone <- m.RunPush(ctx, p, agentcontext.PushOptions{
|
pushDone <- m.RunPush(ctx, p, agentcontext.PushOptions{
|
||||||
Logger: testutil.Logger(t).Named("push"),
|
Logger: testutil.Logger(t).Named("push"),
|
||||||
InitialBackoff: 10 * time.Millisecond,
|
InitialBackoff: time.Second,
|
||||||
|
Clock: mClock,
|
||||||
})
|
})
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// First push hits transient, second succeeds.
|
// First push hits transient and arms the retry timer. Wait for
|
||||||
|
// the timer creation, then advance the clock past the backoff.
|
||||||
|
call := trap.MustWait(ctx)
|
||||||
|
call.MustRelease(ctx)
|
||||||
|
mClock.Advance(time.Second).MustWait(ctx)
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-p.signal:
|
case <-p.signal:
|
||||||
case <-time.After(testutil.WaitShort):
|
case <-time.After(testutil.WaitShort):
|
||||||
|
|||||||
@@ -70,20 +70,6 @@ var skipDirNames = map[string]struct{}{
|
|||||||
"__pycache__": {},
|
"__pycache__": {},
|
||||||
}
|
}
|
||||||
|
|
||||||
// skillsParentNames are directory basenames that signal a
|
|
||||||
// skills container; their immediate children are scanned for
|
|
||||||
// SKILL.md files.
|
|
||||||
var skillsParentNames = map[string]struct{}{
|
|
||||||
"skills": {},
|
|
||||||
".agents": {}, // covers ".agents/skills/<name>"
|
|
||||||
"agents": {},
|
|
||||||
"plugins": {}, // claude code plugin cache layout
|
|
||||||
"cache": {},
|
|
||||||
".coder": {},
|
|
||||||
".claude": {},
|
|
||||||
"skills-dir": {},
|
|
||||||
}
|
|
||||||
|
|
||||||
// recognizedInstructionFile reports whether name is one of the
|
// recognizedInstructionFile reports whether name is one of the
|
||||||
// instruction-file conventions, case-insensitively.
|
// instruction-file conventions, case-insensitively.
|
||||||
func recognizedInstructionFile(name string) bool {
|
func recognizedInstructionFile(name string) bool {
|
||||||
@@ -541,18 +527,11 @@ func excluded(r Resource, reason string) Resource {
|
|||||||
|
|
||||||
// isSkillsContainer reports whether dir is a recognized skills
|
// isSkillsContainer reports whether dir is a recognized skills
|
||||||
// container directory whose immediate children carry SKILL.md
|
// container directory whose immediate children carry SKILL.md
|
||||||
// files.
|
// files. Both bare "skills" and nested "<parent>/skills"
|
||||||
|
// directories qualify (e.g. ".agents/skills",
|
||||||
|
// "plugins/foo/skills").
|
||||||
func isSkillsContainer(dir string) bool {
|
func isSkillsContainer(dir string) bool {
|
||||||
base := filepath.Base(dir)
|
return filepath.Base(dir) == "skills"
|
||||||
_, ok := skillsParentNames[base]
|
|
||||||
if ok && base == "skills" {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
// "<x>/skills" form (e.g. ".agents/skills", "plugins/foo/skills").
|
|
||||||
if strings.HasSuffix(filepath.ToSlash(dir), "/skills") {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// resourceID builds a stable resource ID. Kind plus canonical
|
// resourceID builds a stable resource ID. Kind plus canonical
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ package agentcontext
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"encoding/hex"
|
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
)
|
)
|
||||||
@@ -207,12 +206,6 @@ func ComputeAggregateHash(resources []Resource) [32]byte {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// AggregateHashHex returns the hex-encoded aggregate hash.
|
|
||||||
// Convenience for log lines and HTTP responses.
|
|
||||||
func (s Snapshot) AggregateHashHex() string {
|
|
||||||
return hex.EncodeToString(s.AggregateHash[:])
|
|
||||||
}
|
|
||||||
|
|
||||||
// writeLengthPrefixed writes a uvarint length prefix followed
|
// writeLengthPrefixed writes a uvarint length prefix followed
|
||||||
// by the raw bytes of s.
|
// by the raw bytes of s.
|
||||||
func writeLengthPrefixed(h interface{ Write([]byte) (int, error) }, s string) {
|
func writeLengthPrefixed(h interface{ Write([]byte) (int, error) }, s string) {
|
||||||
|
|||||||
@@ -87,13 +87,3 @@ func TestComputeAggregateHash_ChangesOnContent(t *testing.T) {
|
|||||||
hash3 := agentcontext.ComputeAggregateHash([]agentcontext.Resource{withStatus})
|
hash3 := agentcontext.ComputeAggregateHash([]agentcontext.Resource{withStatus})
|
||||||
require.NotEqual(t, hash1, hash3)
|
require.NotEqual(t, hash1, hash3)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestSnapshotAggregateHashHex(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
snap := agentcontext.Snapshot{
|
|
||||||
AggregateHash: [32]byte{0xde, 0xad, 0xbe, 0xef},
|
|
||||||
}
|
|
||||||
require.Equal(t,
|
|
||||||
"deadbeef0000000000000000000000000000000000000000000000000000000000000000"[:64],
|
|
||||||
snap.AggregateHashHex())
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -33,11 +33,11 @@ func TestWatcher_FiresOnAgentsMdEdit(t *testing.T) {
|
|||||||
ctx := testutil.Context(t, testutil.WaitShort)
|
ctx := testutil.Context(t, testutil.WaitShort)
|
||||||
w.Sync(ctx, []agentcontext.ScanRoot{{Path: dir}})
|
w.Sync(ctx, []agentcontext.ScanRoot{{Path: dir}})
|
||||||
|
|
||||||
// Edit the file. Use a slight delay so fsnotify is ready.
|
// Rewrite the file inside Eventually so the test does not race
|
||||||
time.Sleep(50 * time.Millisecond)
|
// fsnotify's watch-setup window. As soon as the watch is live,
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "AGENTS.md"), []byte("v2"), 0o600))
|
// the next write fires the debounce timer.
|
||||||
|
|
||||||
require.Eventually(t, func() bool {
|
require.Eventually(t, func() bool {
|
||||||
|
_ = os.WriteFile(filepath.Join(dir, "AGENTS.md"), []byte("v2"), 0o600)
|
||||||
return atomic.LoadInt32(&fires) >= 1
|
return atomic.LoadInt32(&fires) >= 1
|
||||||
}, testutil.WaitShort, testutil.IntervalFast, "expected at least one fire after AGENTS.md edit")
|
}, testutil.WaitShort, testutil.IntervalFast, "expected at least one fire after AGENTS.md edit")
|
||||||
}
|
}
|
||||||
@@ -60,12 +60,14 @@ func TestWatcher_FiresOnNewSkillFile(t *testing.T) {
|
|||||||
ctx := testutil.Context(t, testutil.WaitShort)
|
ctx := testutil.Context(t, testutil.WaitShort)
|
||||||
w.Sync(ctx, []agentcontext.ScanRoot{{Path: dir}})
|
w.Sync(ctx, []agentcontext.ScanRoot{{Path: dir}})
|
||||||
|
|
||||||
time.Sleep(50 * time.Millisecond)
|
// Create SKILL.md inside Eventually so the test does not race
|
||||||
|
// fsnotify's watch-setup window. The Manager pre-creates the
|
||||||
|
// skill dir, then rewrites SKILL.md each tick until the watcher
|
||||||
|
// fires at least once.
|
||||||
skillDir := filepath.Join(skillsRoot, "foo")
|
skillDir := filepath.Join(skillsRoot, "foo")
|
||||||
require.NoError(t, os.MkdirAll(skillDir, 0o755))
|
require.NoError(t, os.MkdirAll(skillDir, 0o755))
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(skillDir, "SKILL.md"), []byte("---\nname: foo\ndescription: bar\n---\nbody"), 0o600))
|
|
||||||
|
|
||||||
require.Eventually(t, func() bool {
|
require.Eventually(t, func() bool {
|
||||||
|
_ = os.WriteFile(filepath.Join(skillDir, "SKILL.md"), []byte("---\nname: foo\ndescription: bar\n---\nbody"), 0o600)
|
||||||
return atomic.LoadInt32(&fires) >= 1
|
return atomic.LoadInt32(&fires) >= 1
|
||||||
}, testutil.WaitShort, testutil.IntervalFast, "expected fire after SKILL.md create")
|
}, testutil.WaitShort, testutil.IntervalFast, "expected fire after SKILL.md create")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import (
|
|||||||
// The slice is intentionally tolerant of missing entries; the
|
// The slice is intentionally tolerant of missing entries; the
|
||||||
// resolver silently skips canonicalization failures and
|
// resolver silently skips canonicalization failures and
|
||||||
// non-existent paths.
|
// non-existent paths.
|
||||||
func defaultContextRoots(_ agentcontextconfig.Config, _ func() string) []string {
|
func defaultContextRoots() []string {
|
||||||
roots := make([]string, 0, 8)
|
roots := make([]string, 0, 8)
|
||||||
|
|
||||||
// Working directory is added by the manager itself via the
|
// Working directory is added by the manager itself via the
|
||||||
|
|||||||
Reference in New Issue
Block a user