diff --git a/internal/cache/antigravity_reasoning_replay_cache_test.go b/internal/cache/antigravity_reasoning_replay_cache_test.go index 354aee8b..114ca2da 100644 --- a/internal/cache/antigravity_reasoning_replay_cache_test.go +++ b/internal/cache/antigravity_reasoning_replay_cache_test.go @@ -17,6 +17,7 @@ type fakeAntigravityReasoningReplayKVClient struct { mu sync.Mutex values map[string][]byte expireCount int + casErr error } func newFakeAntigravityReasoningReplayKVClient() *fakeAntigravityReasoningReplayKVClient { @@ -40,6 +41,9 @@ func (c *fakeAntigravityReasoningReplayKVClient) KVSet(_ context.Context, key st func (c *fakeAntigravityReasoningReplayKVClient) KVCompareAndSwap(_ context.Context, key string, expected []byte, expectedExists bool, value []byte, _ time.Duration) (bool, error) { c.mu.Lock() defer c.mu.Unlock() + if c.casErr != nil { + return false, c.casErr + } current, exists := c.values[key] if exists != expectedExists || (exists && !bytes.Equal(current, expected)) { return false, nil @@ -390,6 +394,32 @@ func TestAntigravityReasoningReplayHomeGenerationRejectsSuccessfulValueABA(t *te } } +func TestAntigravityReasoningReplayHomeReportsCASErrors(t *testing.T) { + // The cache layer keeps reporting CAS failures honestly. Deciding that a + // replay failure must not fail the request is the executor's job, so this + // layer must not start swallowing errors. + client := newFakeAntigravityReasoningReplayKVClient() + client.casErr = fmt.Errorf("ERR unknown command 'cas'") + useFakeAntigravityReasoningReplayKVClient(t, client, true) + const model, session = "gemini-3.6-flash-high", "home-cas-error" + + _, _, found, errGet := GetAntigravityReasoningReplayItemsWithSnapshotRequired(context.Background(), model, session) + if errGet == nil { + t.Fatal("GetAntigravityReasoningReplayItemsWithSnapshotRequired() error = nil, want the CAS error") + } + if found { + t.Fatal("GetAntigravityReasoningReplayItemsWithSnapshotRequired() found = true, want false") + } + + snapshot := AntigravityReasoningReplaySnapshot{loaded: true} + if _, errReplace := ReplaceAntigravityReasoningReplayItemsIfUnchanged(context.Background(), model, session, snapshot, [][]byte{antigravityReplayTestItem("home-cas-error-sig-1")}); errReplace == nil { + t.Fatal("ReplaceAntigravityReasoningReplayItemsIfUnchanged() error = nil, want the CAS error") + } + if _, errDelete := DeleteAntigravityReasoningReplayItemsIfUnchanged(context.Background(), model, session, snapshot); errDelete == nil { + t.Fatal("DeleteAntigravityReasoningReplayItemsIfUnchanged() error = nil, want the CAS error") + } +} + func TestAntigravityReasoningReplayHomeCASRetryRejectsOversizedValue(t *testing.T) { client := newFakeAntigravityReasoningReplayKVClient() useFakeAntigravityReasoningReplayKVClient(t, client, true) diff --git a/internal/home/client.go b/internal/home/client.go index 92b8cac5..f4a295c6 100644 --- a/internal/home/client.go +++ b/internal/home/client.go @@ -94,8 +94,24 @@ var ( ErrModelsNotFound = errors.New("home models not found") ErrPluginSyncUnsupported = errors.New("home plugin sync is unsupported") ErrDispatchFenced = errors.New("home auth dispatch is fenced") + // ErrCompareAndSwapUnsupported reports that this Home predates the CAS command. + ErrCompareAndSwapUnsupported = errors.New("home compare-and-swap is unsupported") ) +// isHomeCommandUnsupported reports whether Home rejected a command it does not +// implement. It mirrors isHomeAppLogUnsupported in internal/logging; the two are +// kept separate so the packages stay decoupled. +func isHomeCommandUnsupported(err error) bool { + for err != nil { + message := strings.ToLower(strings.TrimSpace(err.Error())) + if strings.Contains(message, "unknown command") || strings.Contains(message, "unsupported command") { + return true + } + err = errors.Unwrap(err) + } + return false +} + // IsMembershipTakeoverUnavailableError reports whether Home cannot preserve the previous membership state. func IsMembershipTakeoverUnavailableError(err error) bool { if err == nil { @@ -177,6 +193,13 @@ type Client struct { heartbeatOK atomic.Bool dispatchFenced atomic.Bool ambiguousDispatch atomic.Bool + // casUnsupported latches when Home does not implement the CAS command. + // It is deliberately NOT carried across NewLifetime: CAS support is a + // property of the Home deployment, so re-probing once per client lifetime + // lets a Home upgrade take effect on the next reconnect instead of + // requiring a CPA restart. The probe costs one round trip that returns an + // error without performing any write. + casUnsupported atomic.Bool recoveryState atomic.Uint32 instanceID string legacyMembership bool @@ -1033,35 +1056,44 @@ func (c *Client) KVSetNX(ctx context.Context, key string, value []byte, ttl time } // KVCompareAndSwap atomically replaces a value only when its current state matches the expected state. +// +// It uses Home's dedicated CAS command: +// +// CAS [PX ] +// +// Omitting PX stores the value without a TTL. Home replies integer 1 when the +// swap happened and integer 0 when the state did not match. Deployments that +// predate CAS reject the command, which latches ErrCompareAndSwapUnsupported for +// this client lifetime so later calls skip the round trip. func (c *Client) KVCompareAndSwap(ctx context.Context, key string, expected []byte, expectedExists bool, value []byte, ttl time.Duration) (bool, error) { + if c == nil { + return false, ErrNotConnected + } + if c.casUnsupported.Load() { + return false, ErrCompareAndSwapUnsupported + } cmd, errClient := c.commandClient() if errClient != nil { return false, errClient } - const script = ` -local current = redis.call("GET", KEYS[1]) -if ARGV[1] == "1" then - if not current or current ~= ARGV[2] then - return 0 - end -elseif current then - return 0 -end -local ttl = tonumber(ARGV[4]) -if ttl and ttl > 0 then - redis.call("SET", KEYS[1], ARGV[3], "PX", ttl) -else - redis.call("SET", KEYS[1], ARGV[3]) -end -return 1 -` expectedFlag := "0" if expectedExists { expectedFlag = "1" } - result, errEval := cmd.Eval(ctx, script, []string{key}, expectedFlag, expected, value, durationCeil(ttl, time.Millisecond)).Int64() - if errEval != nil { - return false, errEval + args := make([]any, 0, 7) + args = append(args, "CAS", key, expectedFlag, expected, value) + if milliseconds := durationCeil(ttl, time.Millisecond); milliseconds > 0 { + args = append(args, "PX", milliseconds) + } + result, errCAS := cmd.Do(ctx, args...).Int64() + if errCAS != nil { + if isHomeCommandUnsupported(errCAS) { + if c.casUnsupported.CompareAndSwap(false, true) { + log.Warnf("home kv: this Home does not implement the CAS command; Antigravity and Codex reasoning replay are disabled until Home is upgraded") + } + return false, ErrCompareAndSwapUnsupported + } + return false, errCAS } return result == 1, nil } diff --git a/internal/home/client_test.go b/internal/home/client_test.go index 9d0bc69a..66f87d75 100644 --- a/internal/home/client_test.go +++ b/internal/home/client_test.go @@ -534,9 +534,9 @@ func TestKVSetConditionUnmetReturnsFalse(t *testing.T) { } } -func TestKVCompareAndSwapReturnsScriptResult(t *testing.T) { +func TestKVCompareAndSwapSendsCASCommand(t *testing.T) { client, commands := newRedisCommandTestClient(t, func(args []string) string { - if len(args) > 0 && strings.EqualFold(args[0], "EVAL") { + if len(args) > 0 && strings.EqualFold(args[0], "CAS") { return ":1\r\n" } return "-ERR unexpected command\r\n" @@ -549,8 +549,70 @@ func TestKVCompareAndSwapReturnsScriptResult(t *testing.T) { if !swapped { t.Fatal("KVCompareAndSwap() swapped = false, want true") } - if lastCommand := commands.Last(); len(lastCommand) < 2 || !strings.EqualFold(lastCommand[0], "EVAL") { - t.Fatalf("last command = %#v, want EVAL", lastCommand) + want := []string{"CAS", "key", "1", "old", "new", "PX", "1500"} + if lastCommand := commands.Last(); !reflect.DeepEqual(lastCommand, want) { + t.Fatalf("last command = %#v, want %#v", lastCommand, want) + } +} + +func TestKVCompareAndSwapOmitsPXWithoutTTL(t *testing.T) { + client, commands := newRedisCommandTestClient(t, func(args []string) string { + if len(args) > 0 && strings.EqualFold(args[0], "CAS") { + return ":1\r\n" + } + return "-ERR unexpected command\r\n" + }) + + if _, errCAS := client.KVCompareAndSwap(context.Background(), "key", nil, false, []byte("new"), 0); errCAS != nil { + t.Fatalf("KVCompareAndSwap() error = %v", errCAS) + } + // An absent expected value is sent as an empty bulk string, and no TTL means + // no PX, which tells Home to store the value without an expiry. + want := []string{"CAS", "key", "0", "", "new"} + if lastCommand := commands.Last(); !reflect.DeepEqual(lastCommand, want) { + t.Fatalf("last command = %#v, want %#v", lastCommand, want) + } +} + +func TestKVCompareAndSwapReportsMismatch(t *testing.T) { + client, _ := newRedisCommandTestClient(t, func(args []string) string { + if len(args) > 0 && strings.EqualFold(args[0], "CAS") { + return ":0\r\n" + } + return "-ERR unexpected command\r\n" + }) + + swapped, errCAS := client.KVCompareAndSwap(context.Background(), "key", []byte("old"), true, []byte("new"), time.Minute) + if errCAS != nil { + t.Fatalf("KVCompareAndSwap() error = %v", errCAS) + } + if swapped { + t.Fatal("KVCompareAndSwap() swapped = true, want false") + } +} + +func TestKVCompareAndSwapLatchesUnsupportedHome(t *testing.T) { + client, commands := newRedisCommandTestClient(t, func(args []string) string { + if len(args) > 0 && strings.EqualFold(args[0], "CAS") { + return "-ERR unknown command 'cas'\r\n" + } + return "-ERR unexpected command\r\n" + }) + + _, errFirst := client.KVCompareAndSwap(context.Background(), "key", nil, false, []byte("new"), time.Minute) + if !errors.Is(errFirst, ErrCompareAndSwapUnsupported) { + t.Fatalf("KVCompareAndSwap() first error = %v, want ErrCompareAndSwapUnsupported", errFirst) + } + if sent := commands.CountCommandKey("CAS", "key"); sent != 1 { + t.Fatalf("CAS sent %d times, want 1", sent) + } + + _, errSecond := client.KVCompareAndSwap(context.Background(), "key", nil, false, []byte("new"), time.Minute) + if !errors.Is(errSecond, ErrCompareAndSwapUnsupported) { + t.Fatalf("KVCompareAndSwap() second error = %v, want ErrCompareAndSwapUnsupported", errSecond) + } + if sent := commands.CountCommandKey("CAS", "key"); sent != 1 { + t.Fatalf("CAS sent %d times after latching, want 1", sent) } } diff --git a/internal/runtime/executor/antigravity_executor_execute.go b/internal/runtime/executor/antigravity_executor_execute.go index 415914ae..471c9dc3 100644 --- a/internal/runtime/executor/antigravity_executor_execute.go +++ b/internal/runtime/executor/antigravity_executor_execute.go @@ -217,8 +217,8 @@ attemptLoop: } } if errClear := clearAntigravityReasoningReplayOnInvalidSignature(ctx, replayScope, httpResp.StatusCode, bodyBytes); errClear != nil { - err = errClear - return resp, err + // Report the upstream failure rather than the cleanup failure. + logAntigravityReasoningReplayDegraded(replayScope, "invalidate", errClear) } err = newAntigravityStatusErr(httpResp.StatusCode, bodyBytes) return resp, err @@ -452,8 +452,8 @@ attemptLoop: } } if errClear := clearAntigravityReasoningReplayOnInvalidSignature(ctx, replayScope, httpResp.StatusCode, bodyBytes); errClear != nil { - err = errClear - return resp, err + // Report the upstream failure rather than the cleanup failure. + logAntigravityReasoningReplayDegraded(replayScope, "invalidate", errClear) } err = newAntigravityStatusErr(httpResp.StatusCode, bodyBytes) return resp, err diff --git a/internal/runtime/executor/antigravity_executor_stream.go b/internal/runtime/executor/antigravity_executor_stream.go index 4990946b..b90fc84f 100644 --- a/internal/runtime/executor/antigravity_executor_stream.go +++ b/internal/runtime/executor/antigravity_executor_stream.go @@ -225,8 +225,8 @@ attemptLoop: } } if errClear := clearAntigravityReasoningReplayOnInvalidSignature(ctx, replayScope, httpResp.StatusCode, bodyBytes); errClear != nil { - err = errClear - return nil, err + // Report the upstream failure rather than the cleanup failure. + logAntigravityReasoningReplayDegraded(replayScope, "invalidate", errClear) } err = newAntigravityStatusErr(httpResp.StatusCode, bodyBytes) return nil, err diff --git a/internal/runtime/executor/antigravity_reasoning_replay.go b/internal/runtime/executor/antigravity_reasoning_replay.go index 7c8e93a0..51733113 100644 --- a/internal/runtime/executor/antigravity_reasoning_replay.go +++ b/internal/runtime/executor/antigravity_reasoning_replay.go @@ -5,12 +5,14 @@ import ( "context" "crypto/sha256" "encoding/json" + "errors" "fmt" "net/http" "reflect" "strings" internalcache "github.com/router-for-me/CLIProxyAPI/v7/internal/cache" + homekv "github.com/router-for-me/CLIProxyAPI/v7/internal/home" "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" internalsignature "github.com/router-for-me/CLIProxyAPI/v7/internal/signature" "github.com/router-for-me/CLIProxyAPI/v7/internal/util" @@ -230,13 +232,36 @@ func antigravityReasoningReplayResolveContentIndex(payload []byte, cached int) i return -1 } +// logAntigravityReasoningReplayDegraded reports that a replay-state operation +// failed and the request continued without it. A Home that predates the CAS +// command fails every call, and the Home client already warns once about that, +// so those are logged at debug level to avoid one warning per request. +func logAntigravityReasoningReplayDegraded(scope antigravityReasoningReplayScope, stage string, err error) { + if err == nil { + return + } + if errors.Is(err, homekv.ErrCompareAndSwapUnsupported) { + log.Debugf("antigravity executor: reasoning replay %s unavailable on this Home (session=%s): %v", + stage, antigravityReplayLogKey(scope.sessionKey), err) + return + } + log.Warnf("antigravity executor: reasoning replay %s failed; continuing without replay (session=%s): %v", + stage, antigravityReplayLogKey(scope.sessionKey), err) +} + func prepareAntigravityGeminiReasoningReplayPayload(ctx context.Context, modelName string, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, payload []byte) ([]byte, antigravityReasoningReplayScope, error) { if !antigravityUsesReasoningReplayCache(modelName) { return payload, antigravityReasoningReplayScope{}, nil } updated, scope, replayApplied, errReplay := applyAntigravityReasoningReplayCache(ctx, modelName, req, opts, payload) if errReplay != nil { - return payload, scope, errReplay + // Replay state is an optimization, not a correctness requirement: a ledger + // miss is already a tolerated outcome below. Failing the request here would + // surface as an untyped executor error, which MarkResult treats as a + // credential fault and uses to mark every candidate credential unavailable. + // Degrade to "no replay this turn" instead. + logAntigravityReasoningReplayDegraded(scope, "read", errReplay) + updated = payload } updated = normalizeAntigravityGeminiFunctionResponseRoles(updated) if antigravityPayloadHasClaudeToolProvenanceID(updated) { @@ -255,7 +280,9 @@ func prepareAntigravityGeminiReasoningReplayPayload(ctx context.Context, modelNa originalPairingValid := internalsignature.ValidateGeminiFunctionCallPairing(payload) == nil if replayApplied && originalPairingValid && scope.valid() { if _, errDelete := internalcache.DeleteAntigravityReasoningReplayItemsIfUnchanged(ctx, scope.modelName, scope.sessionKey, scope.cacheSnapshot); errDelete != nil { - return payload, scope, errDelete + // Invalidation is best-effort cleanup. Returning it here would replace + // the pairing diagnosis below with an untyped error. + logAntigravityReasoningReplayDegraded(scope, "invalidate", errDelete) } } return payload, scope, statusErr{code: http.StatusBadRequest, msg: fmt.Sprintf("antigravity executor: invalid Gemini function call history: %v", errPairing)} diff --git a/internal/runtime/executor/antigravity_reasoning_replay_test.go b/internal/runtime/executor/antigravity_reasoning_replay_test.go index bfa508eb..7ab99fce 100644 --- a/internal/runtime/executor/antigravity_reasoning_replay_test.go +++ b/internal/runtime/executor/antigravity_reasoning_replay_test.go @@ -8,6 +8,8 @@ import ( "testing" internalcache "github.com/router-for-me/CLIProxyAPI/v7/internal/cache" + "github.com/router-for-me/CLIProxyAPI/v7/internal/config" + homekv "github.com/router-for-me/CLIProxyAPI/v7/internal/home" internalsignature "github.com/router-for-me/CLIProxyAPI/v7/internal/signature" "github.com/router-for-me/CLIProxyAPI/v7/internal/util" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" @@ -49,6 +51,28 @@ func TestAntigravityReasoningReplayAccumulatorMultiToolSSEChunks(t *testing.T) { } } +func TestPrepareAntigravityGeminiReasoningReplayPayloadToleratesHomeKVFailure(t *testing.T) { + // An enabled Home client with no heartbeat makes CurrentKVClient report home + // mode with an error, which is how every Home-side KV failure reaches the + // replay cache — including the "unknown command 'cas'" case from an older + // Home. The request must proceed without replay rather than fail, because a + // bare executor error would make MarkResult mark the credential unavailable. + homekv.SetCurrent(homekv.New(config.HomeConfig{Enabled: true})) + t.Cleanup(func() { homekv.SetCurrent(nil) }) + + payload := []byte(`{"sessionId":"kv-failure","request":{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}}`) + out, _, errPrepare := prepareAntigravityGeminiReasoningReplayPayload(context.Background(), "gemini-3-flash-agent", cliproxyexecutor.Request{}, cliproxyexecutor.Options{}, payload) + if errPrepare != nil { + t.Fatalf("prepare error = %v, want nil so the request proceeds without replay", errPrepare) + } + if len(out) == 0 { + t.Fatal("prepare returned an empty payload") + } + if got := gjson.GetBytes(out, "sessionId").String(); got != "kv-failure" { + t.Fatalf("payload sessionId = %q, want kv-failure", got) + } +} + func TestPrepareAntigravityGeminiReasoningReplayPayloadRejectsToolOutputsAcrossUserBoundary(t *testing.T) { payload := []byte(`{"sessionId":"tool-output-boundary","request":{"contents":[{"role":"model","parts":[{"functionCall":{"id":"call-1","name":"run","args":{}}},{"functionCall":{"id":"call-2","name":"run","args":{}}}]},{"role":"model","parts":[{"functionResponse":{"id":"call-1","name":"run","response":{"result":"one"}}}]},{"role":"user","parts":[{"text":"boundary"}]},{"role":"model","parts":[{"functionResponse":{"id":"call-2","name":"run","response":{"result":"two"}}}]}]}}`) _, _, errPrepare := prepareAntigravityGeminiReasoningReplayPayload(context.Background(), "gemini-3.6-flash-high", cliproxyexecutor.Request{}, cliproxyexecutor.Options{}, payload)