diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 47141463f..518cff4c4 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -20,7 +20,17 @@ import ( ) func TestMain(m *testing.M) { - os.Exit(testenv.RunIsolatedMain(m)) + dir, err := os.MkdirTemp("", "roborev-agy-settings-*") + if err == nil { + antigravitySettingsPathForTest = func() string { + return filepath.Join(dir, ".gemini", "antigravity-cli", "settings.json") + } + } + code := testenv.RunIsolatedMain(m) + if dir != "" { + _ = os.RemoveAll(dir) + } + os.Exit(code) } func TestAgentRegistry(t *testing.T) { diff --git a/internal/agent/antigravity_settings.go b/internal/agent/antigravity_settings.go new file mode 100644 index 000000000..163d1c0ac --- /dev/null +++ b/internal/agent/antigravity_settings.go @@ -0,0 +1,189 @@ +package agent + +import ( + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + "sync" +) + +// Official agy settings path. There is no documented --settings flag or env +// override that headless print-mode honors; permissions are read from +// ~/.gemini/antigravity-cli/settings.json. +// See https://antigravity.google/docs/cli/permissions/ +func defaultAntigravitySettingsPath() string { + home, err := os.UserHomeDir() + if err != nil || home == "" { + return "" + } + return filepath.Join(home, ".gemini", "antigravity-cli", "settings.json") +} + +// antigravitySettingsPathForTest, when set, redirects settings writes away +// from the developer's real ~/.gemini tree (agent tests do not isolate HOME). +var antigravitySettingsPathForTest func() string + +func antigravitySettingsPath() string { + if antigravitySettingsPathForTest != nil { + return antigravitySettingsPathForTest() + } + return defaultAntigravitySettingsPath() +} + +// Inspect commands reviews run (pwd/wc/ls/...) must be allowlisted. In +// headless print mode, unconfigured command() actions default to Ask and +// are soft-denied or hard-fail with "permission check failed for command". +var antigravityReviewAllowPermissions = []string{ + "read_file(*)", + "command(pwd)", + "command(wc)", + "command(ls)", + "command(cat)", + "command(head)", + "command(tail)", + "command(stat)", + "command(file)", +} + +var antigravitySettingsMu sync.Mutex + +// ensureAntigravityReviewPermissions merges the allow-rules non-agentic +// reviews need into settings.json. Existing keys and allow entries are +// preserved; only missing allow strings are appended. Invalid JSON is +// left untouched. +func ensureAntigravityReviewPermissions(settingsPath string) error { + if settingsPath == "" { + return nil + } + + antigravitySettingsMu.Lock() + defer antigravitySettingsMu.Unlock() + + doc := map[string]any{} + raw, err := os.ReadFile(settingsPath) + switch { + case err == nil: + if trimmed := trimSpaceBytes(raw); len(trimmed) > 0 { + if err := json.Unmarshal(raw, &doc); err != nil { + return fmt.Errorf("parse %s: %w", settingsPath, err) + } + } + case os.IsNotExist(err): + // create below + default: + return fmt.Errorf("read %s: %w", settingsPath, err) + } + + permissions, err := settingsObject(doc, "permissions") + if err != nil { + return fmt.Errorf("%s: %w", settingsPath, err) + } + doc["permissions"] = permissions + + allow, changed, err := mergeAllowList(permissions["allow"], antigravityReviewAllowPermissions) + if err != nil { + return fmt.Errorf("%s permissions.allow: %w", settingsPath, err) + } + if !changed && fileExists(settingsPath) { + return nil + } + permissions["allow"] = allow + + if err := writeSettingsJSON(settingsPath, doc); err != nil { + return fmt.Errorf("write %s: %w", settingsPath, err) + } + return nil +} + +func ensureAntigravityReviewSettings() { + path := antigravitySettingsPath() + if path == "" { + log.Printf("antigravity: skipping settings merge; cannot resolve home directory") + return + } + if err := ensureAntigravityReviewPermissions(path); err != nil { + log.Printf("antigravity: could not merge review permissions into %s: %v", path, err) + } +} + +func settingsObject(doc map[string]any, key string) (map[string]any, error) { + raw, ok := doc[key] + if !ok || raw == nil { + return map[string]any{}, nil + } + obj, ok := raw.(map[string]any) + if !ok { + return nil, fmt.Errorf("%s is not a JSON object", key) + } + return obj, nil +} + +func mergeAllowList(existing any, needed []string) (allow []any, changed bool, err error) { + switch v := existing.(type) { + case nil: + allow = nil + case []any: + allow = append([]any(nil), v...) + default: + return nil, false, fmt.Errorf("not a JSON array") + } + + have := make(map[string]struct{}, len(allow)) + for _, item := range allow { + s, ok := item.(string) + if !ok { + continue + } + have[s] = struct{}{} + } + for _, rule := range needed { + if _, ok := have[rule]; ok { + continue + } + allow = append(allow, rule) + have[rule] = struct{}{} + changed = true + } + if existing == nil { + changed = true + } + return allow, changed, nil +} + +func writeSettingsJSON(path string, doc map[string]any) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + data, err := json.MarshalIndent(doc, "", " ") + if err != nil { + return err + } + data = append(data, '\n') + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o644); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +} + +func fileExists(path string) bool { + _, err := os.Stat(path) + return err == nil +} + +func trimSpaceBytes(b []byte) []byte { + i, j := 0, len(b) + for i < j && (b[i] == ' ' || b[i] == '\n' || b[i] == '\r' || b[i] == '\t') { + i++ + } + for j > i && (b[j-1] == ' ' || b[j-1] == '\n' || b[j-1] == '\r' || b[j-1] == '\t') { + j-- + } + return b[i:j] +} diff --git a/internal/agent/antigravity_settings_test.go b/internal/agent/antigravity_settings_test.go new file mode 100644 index 000000000..4024a9d7d --- /dev/null +++ b/internal/agent/antigravity_settings_test.go @@ -0,0 +1,146 @@ +package agent + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestEnsureAntigravityReviewPermissionsCreatesFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "settings.json") + + require.NoError(t, ensureAntigravityReviewPermissions(path)) + + assertSettingsAllow(t, path, antigravityReviewAllowPermissions...) +} + +func TestEnsureAntigravityReviewPermissionsMergesAllow(t *testing.T) { + path := filepath.Join(t.TempDir(), "nested", "settings.json") + writeSettings(t, path, map[string]any{ + "model": "gemini-3.1-pro-preview", + "permissions": map[string]any{ + "allow": []any{"command(git)", "read_file(*)"}, + "deny": []any{"command(rm -rf)"}, + "ask": []any{"command(*)"}, + }, + "enableTerminalSandbox": true, + }) + + require.NoError(t, ensureAntigravityReviewPermissions(path)) + + doc := readSettings(t, path) + assert.Equal(t, "gemini-3.1-pro-preview", doc["model"]) + assert.Equal(t, true, doc["enableTerminalSandbox"]) + + permissions := doc["permissions"].(map[string]any) + assert.Equal(t, []any{"command(rm -rf)"}, permissions["deny"]) + assert.Equal(t, []any{"command(*)"}, permissions["ask"]) + + allow := asStrings(t, permissions["allow"]) + assert.Equal(t, "command(git)", allow[0], "existing allow entries stay first") + assert.Contains(t, allow, "read_file(*)") + assert.Equal(t, 1, countStrings(allow, "read_file(*)"), "do not duplicate existing allows") + for _, rule := range antigravityReviewAllowPermissions { + assert.Contains(t, allow, rule) + } +} + +func TestEnsureAntigravityReviewPermissionsIdempotent(t *testing.T) { + path := filepath.Join(t.TempDir(), "settings.json") + require.NoError(t, ensureAntigravityReviewPermissions(path)) + first := readRaw(t, path) + + require.NoError(t, ensureAntigravityReviewPermissions(path)) + second := readRaw(t, path) + assert.Equal(t, first, second) +} + +func TestEnsureAntigravityReviewPermissionsDoesNotClobberInvalidJSON(t *testing.T) { + path := filepath.Join(t.TempDir(), "settings.json") + require.NoError(t, os.WriteFile(path, []byte("{not-json"), 0o644)) + + err := ensureAntigravityReviewPermissions(path) + require.Error(t, err) + assert.Equal(t, "{not-json", string(readRaw(t, path))) +} + +func TestEnsureAntigravityReviewPermissionsRejectsNonObjectPermissions(t *testing.T) { + path := filepath.Join(t.TempDir(), "settings.json") + writeSettings(t, path, map[string]any{"permissions": "always-proceed"}) + + err := ensureAntigravityReviewPermissions(path) + require.Error(t, err) + doc := readSettings(t, path) + assert.Equal(t, "always-proceed", doc["permissions"]) +} + +func TestEnsureAntigravityReviewPermissionsEmptyPathIsNoop(t *testing.T) { + require.NoError(t, ensureAntigravityReviewPermissions("")) +} + +func TestAntigravityReviewAllowPermissionsCoverProdFailures(t *testing.T) { + assert.Contains(t, antigravityReviewAllowPermissions, "read_file(*)") + for _, cmd := range []string{"pwd", "wc", "ls", "cat", "head", "tail", "stat", "file"} { + assert.Contains(t, antigravityReviewAllowPermissions, "command("+cmd+")") + } +} + +func writeSettings(t *testing.T, path string, doc map[string]any) { + t.Helper() + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o755)) + data, err := json.MarshalIndent(doc, "", " ") + require.NoError(t, err) + require.NoError(t, os.WriteFile(path, append(data, '\n'), 0o644)) +} + +func readSettings(t *testing.T, path string) map[string]any { + t.Helper() + var doc map[string]any + require.NoError(t, json.Unmarshal(readRaw(t, path), &doc)) + return doc +} + +func readRaw(t *testing.T, path string) []byte { + t.Helper() + raw, err := os.ReadFile(path) + require.NoError(t, err) + return raw +} + +func assertSettingsAllow(t *testing.T, path string, want ...string) { + t.Helper() + doc := readSettings(t, path) + permissions, ok := doc["permissions"].(map[string]any) + require.True(t, ok) + allow := asStrings(t, permissions["allow"]) + for _, rule := range want { + assert.Contains(t, allow, rule) + } +} + +func asStrings(t *testing.T, raw any) []string { + t.Helper() + items, ok := raw.([]any) + require.True(t, ok) + out := make([]string, 0, len(items)) + for _, item := range items { + s, ok := item.(string) + require.True(t, ok) + out = append(out, s) + } + return out +} + +func countStrings(items []string, want string) int { + n := 0 + for _, item := range items { + if item == want { + n++ + } + } + return n +} diff --git a/internal/agent/gemini.go b/internal/agent/gemini.go index 5d31dcee1..bba673ac7 100644 --- a/internal/agent/gemini.go +++ b/internal/agent/gemini.go @@ -178,9 +178,10 @@ func (a *GeminiAgent) buildAntigravityArgs(agenticMode bool) []string { if agenticMode { args = append(args, "--dangerously-skip-permissions") - } else { - args = append(args, "--sandbox") } + // Non-agentic print-mode reviews omit --sandbox: agy's sandbox permission + // gate rejects `pwd` (and similar read-only workspace probes) in headless + // print mode. Reviews still do not get --dangerously-skip-permissions. return args } @@ -238,23 +239,33 @@ const ( ) func (a *GeminiAgent) runAntigravity(ctx context.Context, repoPath, prompt string, args []string, output io.Writer) (string, string, error) { + // Headless print mode soft-denies tools that need a confirmation. Merge + // read_file(*) and inspect command() allows into the official settings + // file before launch so reviews emit output. Agentic runs already pass + // --dangerously-skip-permissions and do not need this. + if !a.Agentic { + ensureAntigravityReviewSettings() + } + // Choose the prompt-carrying flag by agy version: >= 1.1.1 takes the prompt - // as the value of --prompt (stdin is ignored); older agy reads it from - // stdin with a bare --print. A bare --print on new agy would swallow the - // following --print-timeout token as the prompt, so the two forms must not - // be mixed. + // as the value of --prompt (stdin is ignored when a prompt flag is present); + // older agy reads it from stdin with a bare --print. A bare --print on new + // agy would swallow the following --print-timeout token as the prompt, so + // the two forms must not be mixed. + // + // Exception: when the prompt exceeds the platform argv cap, omit every + // prompt flag (--prompt/--print/-p). New agy still reads a non-TTY stdin + // as the prompt if no prompt flag is passed (antigravity-cli#582). trimmedPrompt := strings.TrimRight(prompt, "\n") var finalArgs []string var stdin io.Reader if antigravityPromptViaFlag(ctx, a.Command) { - // This contract carries the prompt in argv (agy has no stdin/file - // prompt input here), so bound its length to fail with a clear error - // rather than an opaque exec failure. The ceiling is platform-specific - // (see antigravityMaxPromptArgLen). if size, limit := antigravityPromptArgSize(trimmedPrompt), antigravityMaxPromptArgLen(); size > limit { - return "", "", fmt.Errorf("prompt too large for antigravity argv (size %d, max %d on %s)", size, limit, runtime.GOOS) + finalArgs = append([]string(nil), args...) + stdin = strings.NewReader(trimmedPrompt + "\n") + } else { + finalArgs = append(append([]string(nil), args...), "--prompt", trimmedPrompt) } - finalArgs = append(append([]string(nil), args...), "--prompt", trimmedPrompt) } else { finalArgs = append([]string{"--print"}, args...) stdin = strings.NewReader(trimmedPrompt + "\n") @@ -361,12 +372,13 @@ func utf16CodeUnits(s string) int { } // antigravityMaxPromptArgLen is the ceiling for antigravityPromptArgSize when -// the prompt is passed in argv, the only channel agy print mode offers. The -// limits differ sharply by OS: Windows caps the whole command line at 32767 -// UTF-16 units, Linux caps a single argument at MAX_ARG_STRLEN (128 KiB), and -// macOS only bounds total argv+env (~1 MiB). The default prompt cap (200 KiB) -// exceeds the Linux and Windows ceilings, so a large diff fails with a clear -// error instead of an opaque one. +// the prompt is passed in argv. Prompts larger than this are delivered on +// stdin without a --prompt/--print/-p flag: new agy still reads non-TTY stdin +// when no prompt flag is present (antigravity-cli#582). The limits differ +// sharply by OS: Windows caps the whole command line at 32767 UTF-16 units, +// Linux caps a single argument at MAX_ARG_STRLEN (128 KiB), and macOS only +// bounds total argv+env (~1 MiB). The default prompt cap (200 KiB) exceeds +// the Linux and Windows ceilings. func antigravityMaxPromptArgLen() int { switch runtime.GOOS { case "windows": diff --git a/internal/agent/gemini_test.go b/internal/agent/gemini_test.go index 99760a4cb..a6563ba26 100644 --- a/internal/agent/gemini_test.go +++ b/internal/agent/gemini_test.go @@ -93,27 +93,29 @@ func TestGeminiAntigravityBuildArgs(t *testing.T) { tests := []struct { name string agentic bool - wantFlag string + wantFlags []string unwantedArgs []string }{ { - name: "ReviewMode", - agentic: false, - wantFlag: "--sandbox", + name: "ReviewMode", + agentic: false, + // Print-mode reviews omit --sandbox so `pwd` is not gated, and + // still omit --dangerously-skip-permissions (agentic-only). unwantedArgs: []string{ "--output-format", "--approval-mode", "-m", "--dangerously-skip-permissions", + "--sandbox", // A bare --print would swallow --print-timeout as the // prompt; the prompt is passed via --prompt at run time. "--print", }, }, { - name: "AgenticMode", - agentic: true, - wantFlag: "--dangerously-skip-permissions", + name: "AgenticMode", + agentic: true, + wantFlags: []string{"--dangerously-skip-permissions"}, unwantedArgs: []string{ "--output-format", "--approval-mode", @@ -130,7 +132,9 @@ func TestGeminiAntigravityBuildArgs(t *testing.T) { args := a.buildArgs(tc.agentic) assertFlagValue(t, args, "--print-timeout", "30m") - assert.Contains(t, args, tc.wantFlag) + for _, flag := range tc.wantFlags { + assert.Contains(t, args, flag) + } for _, unwanted := range tc.unwantedArgs { assert.NotContains(t, args, unwanted) } @@ -138,6 +142,49 @@ func TestGeminiAntigravityBuildArgs(t *testing.T) { } } +func TestGeminiAntigravityReviewMergesSettingsAndOmitsYolo(t *testing.T) { + skipIfWindows(t) + + settingsPath := filepath.Join(t.TempDir(), "settings.json") + writeSettings(t, settingsPath, map[string]any{ + "model": "keep-me", + "permissions": map[string]any{ + "allow": []any{"command(git)"}, + "deny": []any{"command(rm -rf)"}, + }, + }) + prev := antigravitySettingsPathForTest + antigravitySettingsPathForTest = func() string { return settingsPath } + t.Cleanup(func() { antigravitySettingsPathForTest = prev }) + + scriptPath := writeTempCommand(t, `#!/bin/sh +if [ "$1" = "--version" ]; then echo "1.1.1"; exit 0; fi +printf '%s\n' "$@" > "$ARGS_FILE" +echo "Review after settings merge" +`) + argsFile := filepath.Join(t.TempDir(), "args") + t.Setenv("ARGS_FILE", argsFile) + a := NewGeminiAgent(scriptPath) + a.Command = filepath.Join(filepath.Dir(scriptPath), "agy") + require.NoError(t, os.Rename(scriptPath, a.Command)) + + res, err := a.Review(context.Background(), t.TempDir(), "sha", "prompt", &bytes.Buffer{}) + require.NoError(t, err) + assert.Equal(t, "Review after settings merge", res) + + assertSettingsAllow(t, settingsPath, "read_file(*)", "command(wc)", "command(pwd)", "command(git)") + doc := readSettings(t, settingsPath) + assert.Equal(t, "keep-me", doc["model"]) + permissions := doc["permissions"].(map[string]any) + assert.Equal(t, []any{"command(rm -rf)"}, permissions["deny"]) + + argsBytes, readErr := os.ReadFile(argsFile) + require.NoError(t, readErr) + argsOut := string(argsBytes) + assert.NotContains(t, argsOut, "--sandbox\n") + assert.NotContains(t, argsOut, "--dangerously-skip-permissions\n") +} + func TestGeminiDetectsAntigravityCommandNames(t *testing.T) { tests := []string{ "agy", @@ -340,6 +387,8 @@ echo "No issues found." argsOut := string(argsBytes) assert.Contains(t, argsOut, "--prompt\nprompt\n") assert.NotContains(t, argsOut, "--print\n") + assert.NotContains(t, argsOut, "--sandbox\n") + assert.NotContains(t, argsOut, "--dangerously-skip-permissions\n") } func TestGeminiAntigravityLegacyStdinContract(t *testing.T) { @@ -471,20 +520,42 @@ func TestUTF16CodeUnits(t *testing.T) { func TestGeminiAntigravityPromptTooLargeForArgv(t *testing.T) { skipIfWindows(t) - // New-contract (flag) path bounds the argv-passed prompt with a clear error. + // Overflowing the platform argv cap must not error. New agy (>= 1.1.1) + // still reads the prompt from non-TTY stdin when no --prompt/--print/-p + // flag is passed (google-antigravity/antigravity-cli#582). scriptPath := writeTempCommand(t, `#!/bin/sh if [ "$1" = "--version" ]; then echo "1.1.1"; exit 0; fi -echo "should not run the review" +cat > "$STDIN_FILE" +printf '%s\n' "$@" > "$ARGS_FILE" +echo "Large prompt review output" +echo "No issues found." `) + stdinFile := filepath.Join(t.TempDir(), "stdin") + argsFile := filepath.Join(t.TempDir(), "args") + t.Setenv("STDIN_FILE", stdinFile) + t.Setenv("ARGS_FILE", argsFile) a := NewGeminiAgent(scriptPath) a.Command = filepath.Join(filepath.Dir(scriptPath), "agy") require.NoError(t, os.Rename(scriptPath, a.Command)) big := strings.Repeat("x", antigravityMaxPromptArgLen()+1) - _, err := a.Review(context.Background(), t.TempDir(), "sha", big, &bytes.Buffer{}) + res, err := a.Review(context.Background(), t.TempDir(), "sha", big, &bytes.Buffer{}) - require.Error(t, err) - assert.Contains(t, err.Error(), "too large for antigravity argv") + require.NoError(t, err) + assert.Equal(t, "Large prompt review output\nNo issues found.", res) + + stdinBytes, readErr := os.ReadFile(stdinFile) + require.NoError(t, readErr) + assert.Equal(t, big+"\n", string(stdinBytes)) + + argsBytes, readErr := os.ReadFile(argsFile) + require.NoError(t, readErr) + argsOut := string(argsBytes) + assert.Contains(t, argsOut, "--print-timeout\n") + assert.NotContains(t, argsOut, "--prompt\n") + assert.NotContains(t, argsOut, "--print\n") + assert.NotContains(t, argsOut, "-p\n") + assert.NotContains(t, argsOut, "--sandbox\n") } func TestGeminiAntigravityVersionProbeFailureDefaultsToPromptFlag(t *testing.T) { diff --git a/internal/daemon/worker.go b/internal/daemon/worker.go index 36d9b6c6b..cd66e80de 100644 --- a/internal/daemon/worker.go +++ b/internal/daemon/worker.go @@ -460,10 +460,7 @@ func (wp *WorkerPool) worker(id int) { // Try to claim a job job, err := wp.db.ClaimJob(workerID) if err != nil { - log.Printf("[%s] Error claiming job: %v", workerID, err) - if wp.errorLog != nil { - wp.errorLog.LogError("worker", fmt.Sprintf("claim job: %v", err), 0) - } + wp.noteClaimError(workerID, err) select { case <-wp.stopCh: log.Printf("[%s] Shutting down", workerID) @@ -505,6 +502,21 @@ func reviewTypeTag(rt string) string { return rt + " " } +// noteClaimError records a ClaimJob failure. SQLITE_BUSY / "database is +// locked" is lock contention: retry/backoff already happened inside +// ClaimJob, so do not spam the daemon error log. Empty-queue is silent +// (nil job, nil error). Real claim failures stay errors. +func (wp *WorkerPool) noteClaimError(workerID string, err error) { + if storage.IsSQLiteBusy(err) { + log.Printf("[%s] Claim job deferred (database busy): %v", workerID, err) + return + } + log.Printf("[%s] Error claiming job: %v", workerID, err) + if wp.errorLog != nil { + wp.errorLog.LogError("worker", fmt.Sprintf("claim job: %v", err), 0) + } +} + func (wp *WorkerPool) processJob(workerID string, job *storage.ReviewJob) { rtTag := reviewTypeTag(job.ReviewType) diff --git a/internal/daemon/worker_claim_test.go b/internal/daemon/worker_claim_test.go new file mode 100644 index 000000000..05f73b7c7 --- /dev/null +++ b/internal/daemon/worker_claim_test.go @@ -0,0 +1,39 @@ +package daemon + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.kenn.io/roborev/internal/storage" +) + +func TestNoteClaimErrorSkipsBusy(t *testing.T) { + el, _ := createTestErrorLog(t) + wp := &WorkerPool{errorLog: el} + + wp.noteClaimError("worker-1", errors.New("database is locked (5) (SQLITE_BUSY)")) + assert.Empty(t, el.Recent(), "SQLITE_BUSY must not spam the daemon error log") + + wp.noteClaimError("worker-1", errors.New("disk I/O error")) + recent := el.Recent() + require.Len(t, recent, 1) + assert.Equal(t, "error", recent[0].Level) + assert.Equal(t, "worker", recent[0].Component) + assert.Contains(t, recent[0].Message, "claim job:") + assert.Contains(t, recent[0].Message, "disk I/O error") +} + +func TestNoteClaimErrorNilErrorLogBusy(t *testing.T) { + wp := &WorkerPool{} + // Must not panic when errorLog is unset (production Start can omit it). + wp.noteClaimError("worker-1", errors.New("database is locked")) + wp.noteClaimError("worker-1", errors.New("boom")) +} + +func TestIsSQLiteBusyMatchesClaimErrors(t *testing.T) { + assert.True(t, storage.IsSQLiteBusy(errors.New("database is locked (5) (SQLITE_BUSY)"))) + assert.False(t, storage.IsSQLiteBusy(errors.New("disk I/O error"))) +} diff --git a/internal/storage/jobs.go b/internal/storage/jobs.go index b97eba36d..425a87d5f 100644 --- a/internal/storage/jobs.go +++ b/internal/storage/jobs.go @@ -428,6 +428,14 @@ func (db *DB) enqueuePanelRunTx(ctx context.Context, exec execer, members []Enqu return memberJobs, synthJob, nil } +// claimJobBusyAttempts / claimJobBusyBackoff cover SQLITE_BUSY after +// the 30s busy_timeout has already elapsed. UPDATE and SELECT are +// retried separately so a successful claim is not repeated. +const ( + claimJobBusyAttempts = 4 + claimJobBusyBackoff = 50 * time.Millisecond +) + // ClaimJob atomically claims the next queued job for a worker. // Jobs whose retry_not_before is in the future are skipped so the retry // backoff applies regardless of which worker happened to fail the prior @@ -442,7 +450,8 @@ func (db *DB) ClaimJob(workerID string) (*ReviewJob, error) { // Atomically claim a job by updating it in a single statement // This prevents race conditions where two workers select the same job - result, err := db.Exec(` + result, err := retryOnSQLiteBusy(claimJobBusyAttempts, claimJobBusyBackoff, time.Sleep, func() (sql.Result, error) { + return db.Exec(` UPDATE review_jobs SET status = 'running', worker_id = ?, started_at = ?, updated_at = ? WHERE id = ( @@ -458,6 +467,7 @@ func (db *DB) ClaimJob(workerID string) (*ReviewJob, error) { WHERE key = ? AND value IN ('true', '1') ) `, workerID, nowStr, nowStr, nowNano, queuePausedStateKey) + }) if err != nil { return nil, err } @@ -474,7 +484,8 @@ func (db *DB) ClaimJob(workerID string) (*ReviewJob, error) { // Now fetch the job we just claimed var job ReviewJob var fields reviewJobScanFields - err = db.QueryRow(` + _, err = retryOnSQLiteBusy(claimJobBusyAttempts, claimJobBusyBackoff, time.Sleep, func() (struct{}, error) { + scanErr := db.QueryRow(` SELECT j.id, j.repo_id, j.commit_id, j.git_ref, j.branch, j.ci_base_branch, j.session_id, j.agent, j.model, j.provider, j.requested_model, j.requested_provider, j.reasoning, j.status, j.enqueued_at, r.root_path, r.name, c.subject, j.diff_content, j.dirty_files, j.prompt, COALESCE(j.agentic, 0), COALESCE(j.prompt_prebuilt, 0), j.job_type, j.review_type, j.output_prefix, j.patch_id, j.parent_job_id, COALESCE(j.worktree_path, ''), j.command_line, COALESCE(j.min_severity, ''), COALESCE(j.backup_agent, ''), COALESCE(j.backup_model, ''), @@ -486,9 +497,11 @@ func (db *DB) ClaimJob(workerID string) (*ReviewJob, error) { ORDER BY j.started_at DESC LIMIT 1 `, workerID).Scan(&job.ID, &job.RepoID, &fields.CommitID, &job.GitRef, &fields.Branch, &fields.CIBaseBranch, &fields.SessionID, &job.Agent, &fields.Model, &fields.Provider, &fields.RequestedModel, &fields.RequestedProvider, &job.Reasoning, &job.Status, &fields.EnqueuedAt, - &job.RepoPath, &job.RepoName, &fields.CommitSubject, &fields.DiffContent, &fields.DirtyFiles, &fields.Prompt, &fields.Agentic, &fields.PromptPrebuilt, &fields.JobType, &fields.ReviewType, - &fields.OutputPrefix, &fields.PatchID, &fields.ParentJobID, &fields.WorktreePath, &fields.CommandLine, &fields.MinSeverity, &fields.BackupAgent, &fields.BackupModel, - &fields.PanelRunUUID, &fields.PanelRole, &fields.PanelName, &fields.PanelMemberName, &fields.PanelMemberIndex, &fields.PanelMemberConfig, &fields.ClaimBlocked, &fields.Source, &job.RetryCount, &fields.UUID) + &job.RepoPath, &job.RepoName, &fields.CommitSubject, &fields.DiffContent, &fields.DirtyFiles, &fields.Prompt, &fields.Agentic, &fields.PromptPrebuilt, &fields.JobType, &fields.ReviewType, + &fields.OutputPrefix, &fields.PatchID, &fields.ParentJobID, &fields.WorktreePath, &fields.CommandLine, &fields.MinSeverity, &fields.BackupAgent, &fields.BackupModel, + &fields.PanelRunUUID, &fields.PanelRole, &fields.PanelName, &fields.PanelMemberName, &fields.PanelMemberIndex, &fields.PanelMemberConfig, &fields.ClaimBlocked, &fields.Source, &job.RetryCount, &fields.UUID) + return struct{}{}, scanErr + }) if err != nil { return nil, err } diff --git a/internal/storage/sqlite_busy.go b/internal/storage/sqlite_busy.go new file mode 100644 index 000000000..d8fdb8523 --- /dev/null +++ b/internal/storage/sqlite_busy.go @@ -0,0 +1,58 @@ +package storage + +import ( + "errors" + "strings" + "time" + + sqlite "modernc.org/sqlite" +) + +// SQLite result codes for lock contention. Named locally so we do not +// depend on modernc.org/sqlite/lib constants. +const ( + sqliteBusy = 5 // SQLITE_BUSY + sqliteLocked = 6 // SQLITE_LOCKED +) + +// IsSQLiteBusy reports whether err is lock contention (SQLITE_BUSY / +// SQLITE_LOCKED), including the "database is locked (5) (SQLITE_BUSY)" +// text modernc.org/sqlite emits after busy_timeout elapses. +func IsSQLiteBusy(err error) bool { + if err == nil { + return false + } + var se *sqlite.Error + if errors.As(err, &se) { + switch se.Code() { + case sqliteBusy, sqliteLocked: + return true + } + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "database is locked") || + strings.Contains(msg, "sqlite_busy") || + strings.Contains(msg, "sqlite_locked") +} + +func retryOnSQLiteBusy[T any](attempts int, backoff time.Duration, sleep func(time.Duration), fn func() (T, error)) (T, error) { + var zero T + if attempts < 1 { + attempts = 1 + } + if sleep == nil { + sleep = time.Sleep + } + var lastErr error + for i := 0; i < attempts; i++ { + v, err := fn() + if err == nil || !IsSQLiteBusy(err) { + return v, err + } + lastErr = err + if i+1 < attempts { + sleep(backoff << i) + } + } + return zero, lastErr +} diff --git a/internal/storage/sqlite_busy_test.go b/internal/storage/sqlite_busy_test.go new file mode 100644 index 000000000..0527d9b61 --- /dev/null +++ b/internal/storage/sqlite_busy_test.go @@ -0,0 +1,97 @@ +package storage + +import ( + "database/sql" + "errors" + "fmt" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestIsSQLiteBusy(t *testing.T) { + assert.False(t, IsSQLiteBusy(nil)) + assert.False(t, IsSQLiteBusy(errors.New("disk I/O error"))) + assert.False(t, IsSQLiteBusy(errors.New("no such table: review_jobs"))) + assert.True(t, IsSQLiteBusy(errors.New("database is locked (5) (SQLITE_BUSY)"))) + assert.True(t, IsSQLiteBusy(fmt.Errorf("claim job: %w", errors.New("SQLITE_BUSY")))) + assert.True(t, IsSQLiteBusy(errors.New("database is locked"))) + assert.True(t, IsSQLiteBusy(errors.New("SQLITE_LOCKED"))) +} + +func TestIsSQLiteBusyRealError(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "busy.db") + dsn := path + "?_pragma=journal_mode(DELETE)&_pragma=busy_timeout(0)" + + holder, err := sql.Open("sqlite", dsn) + require.NoError(t, err) + t.Cleanup(func() { _ = holder.Close() }) + _, err = holder.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY); INSERT INTO t (id) VALUES (1)`) + require.NoError(t, err) + tx, err := holder.Begin() + require.NoError(t, err) + t.Cleanup(func() { _ = tx.Rollback() }) + _, err = tx.Exec(`UPDATE t SET id = 1`) + require.NoError(t, err) + + contender, err := sql.Open("sqlite", dsn) + require.NoError(t, err) + t.Cleanup(func() { _ = contender.Close() }) + _, err = contender.Exec(`UPDATE t SET id = 2`) + require.Error(t, err) + assert.True(t, IsSQLiteBusy(err), "got %v", err) +} + +func TestRetryOnSQLiteBusySucceedsAfterContention(t *testing.T) { + n := 0 + var sleeps []time.Duration + job, err := retryOnSQLiteBusy(4, 50*time.Millisecond, func(d time.Duration) { + sleeps = append(sleeps, d) + }, func() (*ReviewJob, error) { + n++ + if n < 3 { + return nil, errors.New("database is locked (5) (SQLITE_BUSY)") + } + return &ReviewJob{ID: 7}, nil + }) + require.NoError(t, err) + require.NotNil(t, job) + assert.Equal(t, int64(7), job.ID) + assert.Equal(t, 3, n) + assert.Equal(t, []time.Duration{50 * time.Millisecond, 100 * time.Millisecond}, sleeps) +} + +func TestRetryOnSQLiteBusyDoesNotRetryPermanentErrors(t *testing.T) { + n := 0 + _, err := retryOnSQLiteBusy(4, time.Millisecond, func(time.Duration) { + t.Fatal("should not sleep on permanent errors") + }, func() (*ReviewJob, error) { + n++ + return nil, errors.New("disk I/O error") + }) + require.EqualError(t, err, "disk I/O error") + assert.Equal(t, 1, n) +} + +func TestRetryOnSQLiteBusyGivesUp(t *testing.T) { + n := 0 + _, err := retryOnSQLiteBusy(3, time.Millisecond, func(time.Duration) {}, func() (*ReviewJob, error) { + n++ + return nil, errors.New("database is locked (5) (SQLITE_BUSY)") + }) + require.Error(t, err) + assert.True(t, IsSQLiteBusy(err)) + assert.Equal(t, 3, n) +} + +func TestRetryOnSQLiteBusyNilIsSuccess(t *testing.T) { + job, err := retryOnSQLiteBusy(4, time.Millisecond, nil, func() (*ReviewJob, error) { + return nil, nil + }) + require.NoError(t, err) + assert.Nil(t, job) +}