From 20784c67ff520fbc4f88116582f17d9f39939246 Mon Sep 17 00:00:00 2001 From: sususu Date: Tue, 28 Jul 2026 23:20:56 +0800 Subject: [PATCH] fix(home): use Home's CAS command instead of EVAL for replay compare-and-swap Client.KVCompareAndSwap sent Redis EVAL with a Lua script, but the Home RESP subset does not implement EVAL, so Home replied "ERR unknown command 'eval'". That broke the Antigravity and Codex reasoning replay caches in Home mode. Switch the transport to Home's dedicated CAS command: CAS [PX ] The semantics match the old script argument for argument, so KVCompareAndSwap's signature and all its callers are unchanged. Omitting PX when ttl <= 0 mirrors the script's SET-without-PX branch, which clears the TTL. Deployments that predate CAS reject the command. Detect that by matching the unsupported-command error, latch ErrCompareAndSwapUnsupported for the client lifetime so later calls skip the round trip, and warn exactly once. The latch is deliberately not carried across NewLifetime, so a Home upgrade takes effect on the next reconnect rather than requiring a CPA restart. Also stop replay-state failures from failing the request. A bare replay error has no HTTP status, so resultErrorFromError does not classify it as request-scoped and MarkResult marks the credential unavailable for that model, walking every candidate credential until alias resolution has nothing left and returns 503. A ledger miss is already a tolerated outcome, so degrade to "no replay this turn" instead. The pairing failure still returns its 400. Verified end to end against a real Home over RESP/mTLS on PostgreSQL with Antigravity OAuth credentials: patched Home recreates the replay row through CAS with its TTL, while a pre-CAS Home latches once, keeps returning 200 instead of 503, and records no credential error attributable to the replay path. Refs router-for-me/CLIProxyAPIHome#79 --- ...antigravity_reasoning_replay_cache_test.go | 30 ++++++++ internal/home/client.go | 72 +++++++++++++------ internal/home/client_test.go | 70 ++++++++++++++++-- .../executor/antigravity_executor_execute.go | 8 +-- .../executor/antigravity_executor_stream.go | 4 +- .../executor/antigravity_reasoning_replay.go | 31 +++++++- .../antigravity_reasoning_replay_test.go | 24 +++++++ 7 files changed, 207 insertions(+), 32 deletions(-) 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) -- 2.51.2