feat(relay): implement HTTP replay mechanism for OpenAI Responses API (#92) - #107
Merged
Conversation
…#92) 实现本地 HTTP replay 机制，解决上游中转站拒绝 previous_response_id 的兼容性问题。 ## 背景 当 OpenAI Responses HTTP 请求携带 previous_response_id 时，部分上游中转站返回： previous_response_id is only supported on Responses WebSocket v2 同时这些上游的 WebSocket 升级路径不可用（426/101 handshake 失败）。 ## 实现 1. **本地状态存储** (responses_replay_store.go) - 使用 sync.Map 作为进程内存储，key 格式：apiKeyID:groupID:requestModel:hash(responseID) - 每次 HTTP /v1/responses 成功后保存 replay 状态（渠道、key、replay window） - 支持 TTL 自动过期，默认继承 group.SessionKeepTime 2. **请求转换** (relay.go) - 检测到 previous_response_id 时，从本地加载上一次状态 - 调用 BuildReplayRequest 转为自包含形式（合并历史 + 移除 previous_response_id） - 优先复用上一次成功的渠道/key（通过 NewIteratorWithPreference） 3. **可观测性** - 记录 ws_mode=replay 和 ws_recovery=replay 便于排障 - Debug 日志记录状态加载、转换、路由、保存全流程 ## 兼容性 - 与 PR #106 的 item_reference 修复兼容（MarshalResponsesInputItems 现在正确生成 id 和 item_reference） - 复用现有 wsConversationState 和 replay 逻辑，无需数据库迁移 - 不影响 WebSocket 原生续接和 passthrough 模式 ## 测试覆盖 - 单元测试：状态存储、加载、TTL、隔离性 - 集成测试：完整 replay 流程、流式请求、不同 group 隔离、tool calls Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
修复代码审查中发现的关键问题： ## High Priority 修复 1. **多轮 replay 状态持久化** - 移除 !IsOpenAIExactReplayRequest() 的保存限制 - exact replay 成功后继续保存新状态（基于已有状态累积） - 支持连续多轮 HTTP replay (resp1 -> resp2 -> resp3) 2. **容量控制与主动清理** - 增加最大条目数限制 (10,000) - 增加最大内存限制 (100MB) - 后台定时清理过期条目 (5分钟间隔) - 状态大小估算与容量检查 - 增加统计指标 (entries, totalSize) 3. **历史合并失败回退** - 验证 BuildReplayRequest 是否成功生成 RawInputItems - 合并失败时保留原始 previous_response_id - 放弃本地 replay，允许回退到原生续接 - 增加 warn 日志记录失败原因 ## Medium Priority 改进 4. **增强错误日志** - GetInternalResponse 失败时记录 debug 日志 - 历史合并失败记录 warn 日志 - 容量超限记录 warn 日志 5. **降低碰撞风险** - SHA256 hash 截断从 16 hex (64-bit) 增加到 32 hex (128-bit) ## 测试覆盖 6. **新增测试** - TestHTTPReplayMultiTurnChain: 验证连续 3 轮 replay - TestHTTPReplayFailedMergeKeepsOriginalRequest: 验证合并失败时的回退 ## 影响范围 - 修复后支持稳定的多轮 HTTP replay continuation - 防止内存泄漏和无限增长 - 提升可观测性和故障排查能力 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
修复第二轮代码审查中发现的所有关键问题。
## High Priority 修复
1. **流式响应保存问题** (High 1)
- 优先使用 `metrics.InternalResponse`（已由 collectResponse 填充）
- 避免二次调用 `GetInternalResponse()` 消耗 streaming 聚合器
- 确保流式请求的 replay state 能正确保存
2. **统计一致性问题** (High 2)
- 使用 `Swap()` 原子操作保证统计与 map 一致
- 所有 Delete 路径统一更新 entries/totalSize
- `resetResponsesReplayStore()` 同时重置统计
- 容量检查在 Swap 后执行，允许更新已有 key
## Medium Priority 改进
3. **容量检查逻辑** (Medium 1)
- 更新已有 key 时只计算 size delta，不检查 entries 上限
- 新 key 才检查容量，超限时回滚
4. **测试改进** (Medium 4)
- 修复 `TestHTTPReplayFailedMergeKeepsOriginalRequest`
- 使用真正会失败的场景（空 Messages + 空 RawInputItems）
- 验证 relay.go 的 fallback 校验逻辑
5. **代码格式** (Low 1)
- 运行 gofmt 格式化 responses_replay_store.go
## 技术细节
### 统计一致性实现
```go
old, loaded := responsesReplayStore.Swap(key, newEntry)
if loaded {
// 更新：只调整 size 差值
responsesReplayStoreStats.totalSize.Add(delta)
} else {
// 新增：增加 entries，检查容量，超限则回滚
currentEntries := responsesReplayStoreStats.entries.Add(1)
if overflow { Delete + rollback stats }
}
```
### 流式安全保存
```go
// 优先使用已收集的响应（streaming 安全）
internalResponse := metrics.InternalResponse
if internalResponse == nil {
// fallback 到 GetInternalResponse
}
```
## 测试结果
- ✅ 全部 replay 测试通过
- ✅ go test -race 无数据竞争（replay 部分）
- ✅ gofmt 检查通过
## 未修复项（需后续处理）
1. **BuildReplayRequest 语义问题** (Medium 3)
- 当前在 relay.go 调用侧做校验
- 建议后续重构为 BuildReplayRequestStrict
2. **Handler 级端到端测试** (Medium 4)
- 需 mock gin/inbound/outbound
- 建议后续单独 PR 补充
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…nt deletion race
修复第三轮审查发现的并发安全问题。
## 问题
在高并发场景下，replay store 的 Delete 操作可能误删并发写入的新 entry：
1. **loadResponsesReplayState()**: goroutine A 判断 entry 过期，准备 Delete(key)
2. **同时**: goroutine B 对同一 key Swap 进新的有效 entry
3. **误删**: goroutine A 执行 Delete(key)，误删了 B 刚写入的新 entry
同样的问题存在于：
- sweepExpiredResponsesReplayStates() 删除过期 entry
- storeResponsesReplayState() 容量回滚删除
## 修复
使用 `sync.Map.CompareAndDelete(key, oldValue)` 替代 `Delete(key)`：
- 只删除我们检查过的那个 entry
- 如果 key 的 value 已被并发修改，删除失败，避免误删
- 统计更新仅在删除成功后执行
## 代码位置
- loadResponsesReplayState(): L106-108, L112-115
- sweepExpiredResponsesReplayStates(): L63-66, L70-74
- storeResponsesReplayState(): L168-171
## 技术细节
```go
// Before (有竞态)
if entry.expiresAt.After(now) {
responsesReplayStore.Delete(key) // 可能误删并发写入的新 entry
responsesReplayStoreStats.entries.Add(-1)
}
// After (并发安全)
if entry.expiresAt.After(now) {
if responsesReplayStore.CompareAndDelete(key, entry) { // 只删除这个 entry
responsesReplayStoreStats.entries.Add(-1) // 仅删除成功时更新
}
}
```
## 测试结果
✅ go test -race -count=1 -run 'TestHTTPReplay|TestResponsesReplay' ./internal/relay
✅ 所有 replay 测试通过
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
修复 BuildReplayRequest 的危险语义，merge 失败时返回 nil 而非部分转换的请求。
## 问题
原实现中，即使历史合并失败（buildReplayRawInputItems 返回 ok=false），
BuildReplayRequest 仍会返回一个：
- 已移除 previous_response_id
- 标记为 exact replay
- 但缺少 RawInputItems
的请求对象。这对未来的调用方是危险的。
## 修复
BuildReplayRequest 现在：
1. 检查 merge 是否成功（ok && len(mergedRawInputItems) > 0）
2. 失败时直接返回 nil
3. 成功时才返回转换后的请求
调用方（relay.go）简化为：
```go
if replayed := state.BuildReplayRequest(req); replayed != nil {
internalRequest = replayed // 使用转换后的请求
} else {
responsesReplayState = nil // merge 失败，保留原始请求
}
```
## 好处
- **函数语义更清晰**: nil = 失败，非 nil = 成功
- **防止误用**: 未来新调用点无需额外校验 RawInputItems
- **简化调用侧**: relay.go 不再需要检查 len(replayed.OpenAIRawInputItems())
## 代码位置
- ws_session.go:161-189 - BuildReplayRequest 函数本体
- relay.go:97-112 - 调用侧简化
- responses_replay_integration_test.go:425-461 - 测试更新
## 测试结果
✅ 所有 replay 测试通过
✅ go test -race 无数据竞争
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughWalkthroughAdds an in-memory replay store for OpenAI Responses HTTP requests, integrates replay resolution and sticky routing into the relay handler, saves updated state after successful requests, and makes replay request construction return ChangesHTTP Responses Replay
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
internal/relay/responses_replay_store_test.go (1)
198-212: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winStrengthen clone isolation coverage.
This only mutates top-level fields on the original. A shallow copy of
ReplayWindowItems,Transcript, orReplayAliaseswould still pass and could let callers corrupt stored replay state afterloadResponsesReplayState.Suggested direction
// Mutate original state.LastResponseID = "resp_mutated" state.ChannelID = 999 + state.ReplayWindowItems[0] = '{' + if state.Transcript[0].Content.Content != nil { + *state.Transcript[0].Content.Content = "mutated" + } + state.ReplayAliases[0] = "mutated" @@ if loaded.ChannelID != 5 { t.Fatalf("expected cloned state with ChannelID=5, got %d", loaded.ChannelID) } + if string(loaded.ReplayWindowItems) != `[{"type":"message"}]` { + t.Fatalf("expected ReplayWindowItems to be cloned, got %s", loaded.ReplayWindowItems) + } + if loaded.Transcript[0].Content.Content == nil || *loaded.Transcript[0].Content.Content != "hello" { + t.Fatalf("expected Transcript to be cloned, got %+v", loaded.Transcript) + } + if loaded.ReplayAliases[0] != "alias1" { + t.Fatalf("expected ReplayAliases to be cloned, got %+v", loaded.ReplayAliases) + } + + loaded.ReplayAliases[0] = "loaded-mutated" + reloaded := loadResponsesReplayState(30, 300, "gpt-4", "resp_original") + if reloaded == nil || reloaded.ReplayAliases[0] != "alias1" { + t.Fatalf("expected loadResponsesReplayState to return an isolated clone, got %+v", reloaded) + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/relay/responses_replay_store_test.go` around lines 198 - 212, The clone isolation test for loadResponsesReplayState only verifies top-level fields, so it can miss shallow-copy bugs in nested data. Update the test around loadResponsesReplayState to also mutate nested structures on the original state, such as ReplayWindowItems, Transcript, and ReplayAliases, then assert the loaded clone still preserves the original nested values. Keep the focus on the loadResponsesReplayState helper and the replay state fields it copies.internal/relay/responses_replay_integration_test.go (1)
444-451: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover the relay fallback path here. This test only asserts
BuildReplayRequest(...)returnsnil; add a path throughrelay.gothat proves the original request is preserved when replay merge fails, so the handler fallback stays covered.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/relay/responses_replay_integration_test.go` around lines 444 - 451, The current test only covers loadedState.BuildReplayRequest returning nil, but it does not exercise the relay fallback behavior. Update the replay integration test to route the request through the relay.go handling path that consumes BuildReplayRequest, and assert that when merge fails the original request is preserved and used unchanged. Keep the focus on the interaction between loadedState.BuildReplayRequest and the relay handler fallback so the nil case in relay.go is covered end-to-end.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/relay/responses_replay_integration_test.go`:
- Around line 384-401: The current turn 3 checks only confirm that some
transcript exists and that exact replay is enabled, but they do not prove turn 2
history is still preserved. Update the integration test around
loadedState3.BuildReplayRequest and replayedReq3 to assert against
replayedReq3.OpenAIRawInputItems() or the stored ReplayWindowItems so the chain
explicitly verifies that turn 2 survives into turn 3, not just that earlier
state exists.
- Around line 236-250: The replay test fixture in InternalLLMResponse is using
the wrong raw output shape for the assistant turn. Update the
RawResponsesOutputItems in responses_replay_integration_test.go for firstResp so
it represents the assistant’s tool request with a function_call item, matching
the ToolCalls entry on the Message and the replay shape persisted by the relay
after a tool invocation. Keep the existing identifiers like firstResp,
InternalLLMResponse, and ToolCall/GetWeather setup, but replace the client-style
function_call_output payload with the assistant-side function_call
representation.
In `@internal/relay/responses_replay_store.go`:
- Around line 159-183: The capacity check in responsesReplayStore.Put should
also apply when Swap replaces an existing entry, not only for new keys. In the
loaded branch, after updating responsesReplayStoreStats.totalSize for the new
responsesReplayStateEntry, verify the store still fits within
responsesReplayStoreMaxSize and roll back the replacement if it exceeds the
limit. Use the existing Swap, CompareAndDelete, and responsesReplayStoreStats
bookkeeping to restore the old entry and keep entries/totalSize consistent.
- Around line 186-195: The size estimate in estimateStateSize is using fixed
multipliers for Transcript and ReplayAliases, which undercounts real payload
size and can let large states bypass the store limit. Update estimateStateSize
to sum the actual byte sizes of the transcript entries and alias contents (and
any nested fields they contain), using the wsConversationState fields directly
instead of len(...) * constants, so the limit tracks true memory usage more
closely.
---
Nitpick comments:
In `@internal/relay/responses_replay_integration_test.go`:
- Around line 444-451: The current test only covers
loadedState.BuildReplayRequest returning nil, but it does not exercise the relay
fallback behavior. Update the replay integration test to route the request
through the relay.go handling path that consumes BuildReplayRequest, and assert
that when merge fails the original request is preserved and used unchanged. Keep
the focus on the interaction between loadedState.BuildReplayRequest and the
relay handler fallback so the nil case in relay.go is covered end-to-end.
In `@internal/relay/responses_replay_store_test.go`:
- Around line 198-212: The clone isolation test for loadResponsesReplayState
only verifies top-level fields, so it can miss shallow-copy bugs in nested data.
Update the test around loadResponsesReplayState to also mutate nested structures
on the original state, such as ReplayWindowItems, Transcript, and ReplayAliases,
then assert the loaded clone still preserves the original nested values. Keep
the focus on the loadResponsesReplayState helper and the replay state fields it
copies.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ca1268ad-185c-4e3e-9a16-2d95f441522c
📒 Files selected for processing (5)
internal/relay/relay.gointernal/relay/responses_replay_integration_test.gointernal/relay/responses_replay_store.gointernal/relay/responses_replay_store_test.gointernal/relay/ws_session.go
|
大佬什么时候发布新版 |
…test coverage - Add capacity rollback when Swap replacement exceeds size limit - Replace fixed-multiplier estimateStateSize with per-field traversal - Fix RawResponsesOutputItems to use function_call instead of function_call_output - Assert multi-turn chain preserves all turns in replayed RawInputItems - Strengthen clone isolation test with nested structure mutations Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Hureru
added
the
build
mingtian886
approved these changes
Jun 30, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
功能说明
实现 OpenAI Responses API 的 HTTP Replay 机制，解决上游中转站拒绝
previous_response_id的兼容性问题。当 HTTP 请求携带
previous_response_id时，从本地加载上一次成功的 replay 状态，转换为自包含请求（合并历史，移除 previous_response_id），优先路由到成功的 channel/key。核心功能
apiKeyID:groupID:requestModel:hash(responseID)实现细节
状态管理
并发安全
Swap()保证写入与统计的原子性CompareAndDelete()防止误删并发写入的 entry可观测性
ws_mode=replay和ws_recovery=replay代码审查修复
经过三轮专业代码审查和五次迭代修复：
第一轮
第二轮
第三轮
测试覆盖
16 个测试用例，覆盖：
兼容性
变更统计
待改进项（不阻塞合并）
Closes #92
Co-Authored-By: Claude Opus 4.6 (1M context) noreply@anthropic.com
Summary by CodeRabbit
New Features
Bug Fixes
Tests