diff --git a/README.md b/README.md index a94dbe1..2f3640a 100644 --- a/README.md +++ b/README.md @@ -66,10 +66,10 @@ CLIProxyAPI 负责 HTTP 接入、协议转换、上游凭证和实际请求执 ## 当前兼容目标 -- CLIProxyAPI 源码：`.externals/CLIProxyAPI/`，检查时 revision 为 `f43aad7637ad813745bf7d341acb5663617570c5` +- CLIProxyAPI 基线：`v7.2.132` / `78f0c4079e3e6273d65d03b5549cffc898703264`，需依次应用 `patch/` 中两项宿主补丁 - 插件版本：`0.1.0` - Native ABI：`1` -- RPC schema：最高 `3`，注册时按宿主版本向下协商 +- RPC schema：最高 `4`，注册时按宿主版本向下协商；schema v4 提供 Usage 请求身份 - 插件 ID / 动态库文件名：`billing` - 已声明能力：`frontend_auth_provider`（独占）、`scheduler`、`request_interceptor`、`request_lifecycle_plugin`、`usage_plugin`、`management_api` diff --git a/docs/modules/usage-and-statistics.md b/docs/modules/usage-and-statistics.md index 9d9b57a..362336e 100644 --- a/docs/modules/usage-and-statistics.md +++ b/docs/modules/usage-and-statistics.md @@ -26,20 +26,21 @@ CLIProxyAPI 会通过两个独立回调提供请求信息： | 事实 | 来源 | 主要内容 | | --- | --- | --- | | 请求终态 | `request.complete` | Request ID、开始与结束时间、成功/失败/拒绝/取消、状态码和错误 | -| 最终用量 | `usage.handle` | 实际上游、模型、Token、延迟和 Usage 结果 | +| 最终用量 | `usage.handle` | Request ID、Trace ID、实际上游、模型、Token、延迟和 Usage 结果 | 这两个回调可能乱序、重复或只到达其中一个，因此数据库分别保存原始事实，再建立请求明细查询投影。管理台看到的一行是查询结果，不会为了合并展示而修改原始事实或计费账目。 ## 请求关联与独立用量 - Request ID 用于关联一次下游请求的生命周期。 -- 当前目标 CPA 的 Usage 契约不提供 Request ID 或 Execution ID；Request ID 和 Trace ID 来自独立的请求终态。 +- RPC schema v4 的 Usage 契约提供 Request ID 和 Trace ID，用于与请求终态精确关联。 +- 旧 schema 或历史 Usage 没有请求身份时，仍使用保守的唯一时间窗口匹配作为兼容路径。 - 每条可区分的 Usage 事实分别保存，不会为了得到一条整齐记录而把多个用量相加。 - 当前契约无法保证把每次上游重试稳定标记为某个 Execution ID，因此管理台不声明这种保证。 - 没有 Usage 的拒绝、取消或失败请求仍然显示请求终态。 - 只有 Usage、暂时没有终态的记录也可以单独显示，终态到达后投影会自动更新。 -系统仅在模型和请求时间足够接近且匹配关系唯一时，将 Usage 与终态合并；存在并发歧义时宁可保留为两条，也不会错误关联到其他用户的请求。 +schema v4 记录按 Request ID 精确归并，不受同模型并发影响。只有旧 schema 或历史孤立记录才使用模型和请求时间匹配；存在并发歧义时宁可保留为两条，也不会错误关联到其他用户的请求。 ## 请求结果 diff --git a/internal/plugin/app.go b/internal/plugin/app.go index 201be3c..e49780d 100644 --- a/internal/plugin/app.go +++ b/internal/plugin/app.go @@ -178,6 +178,8 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) { } // CPA wire 类型只存在于适配层，采集模块接收与协议无关的观察值。 observed := collection.Record{ + RequestID: record.RequestID, + TraceID: record.TraceID, RequestedAt: record.RequestedAt, APIKey: record.APIKey, AuthID: record.AuthID, diff --git a/internal/plugin/app_test.go b/internal/plugin/app_test.go index feab213..3732665 100644 --- a/internal/plugin/app_test.go +++ b/internal/plugin/app_test.go @@ -95,7 +95,7 @@ func TestUsageRecordWireFieldsMatchTargetContract(t *testing.T) { t.Fatal(err) } want := []string{ - "Provider", "ExecutorType", "Model", "Alias", "APIKey", "AuthID", "AuthIndex", "AuthType", + "RequestID", "TraceID", "Provider", "ExecutorType", "Model", "Alias", "APIKey", "AuthID", "AuthIndex", "AuthType", "Source", "ReasoningEffort", "ServiceTier", "Generate", "RequestedAt", "Latency", "TTFT", "Failed", "Failure", "Detail", "ResponseHeaders", } @@ -109,6 +109,38 @@ func TestUsageRecordWireFieldsMatchTargetContract(t *testing.T) { } } +func TestUsageRecordIdentityReconcilesLifecycleExactly(t *testing.T) { + app := NewApp() + if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: false\n"))); err != nil { + t.Fatal(err) + } + defer app.Shutdown() + + startedAt := time.Now().UTC() + completionRaw, _ := json.Marshal(RequestCompletion{ + RequestID: "request-identity", TraceID: "trace-identity", Model: "deepseek-v4-flash", + StartedAt: startedAt, CompletedAt: startedAt.Add(time.Second), Outcome: RequestCompletionSucceeded, + }) + if _, err := app.HandleMethod(MethodRequestComplete, completionRaw); err != nil { + t.Fatal(err) + } + usageRaw, _ := json.Marshal(UsageRecord{ + RequestID: "request-identity", TraceID: "trace-identity", Provider: "openai", Model: "deepseek-v4-flash", + RequestedAt: startedAt.Add(5 * time.Millisecond), Detail: UsageDetail{InputTokens: 10, OutputTokens: 2, TotalTokens: 12}, + }) + if _, err := app.HandleMethod(MethodUsageHandle, usageRaw); err != nil { + t.Fatal(err) + } + + records, err := app.store.ListRecent(context.Background(), 10) + if err != nil { + t.Fatal(err) + } + if len(records) != 1 || records[0].RequestID != "request-identity" || records[0].TraceID != "trace-identity" || records[0].TotalTokens != 12 { + t.Fatalf("records = %+v", records) + } +} + func TestReconfigureIsAtomicAndFiltersUsage(t *testing.T) { app := NewApp() if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: true\n"))); err != nil { diff --git a/internal/plugin/types.go b/internal/plugin/types.go index 0d38281..6abad60 100644 --- a/internal/plugin/types.go +++ b/internal/plugin/types.go @@ -9,7 +9,7 @@ import ( const ( ABIVersion uint32 = 1 - SchemaVersion uint32 = 3 + SchemaVersion uint32 = 4 PluginName = "billing" Version = "0.1.0" ) @@ -226,6 +226,8 @@ type ManagementResponse struct { // UsageRecord mirrors CLIProxyAPI sdk/pluginapi. Keep it in sync with the // target host because exported field names are part of the JSON wire contract. type UsageRecord struct { + RequestID string + TraceID string Provider string ExecutorType string Model string diff --git a/internal/repository/sqlite_projection.go b/internal/repository/sqlite_projection.go index 3cd8f4e..7522723 100644 --- a/internal/repository/sqlite_projection.go +++ b/internal/repository/sqlite_projection.go @@ -166,7 +166,7 @@ WHERE u.id=?`, usageID) } func syncRequestProjection(ctx context.Context, tx *sql.Tx, requestID string) (bool, error) { - result, err := tx.ExecContext(ctx, ` + _, err := tx.ExecContext(ctx, ` UPDATE request_detail_index SET lifecycle_request_id=?, request_id=?, managed_key_id=COALESCE(NULLIF((SELECT managed_key_id FROM usage_records WHERE id=request_detail_index.usage_id), ''), @@ -180,8 +180,15 @@ WHERE usage_id IN (SELECT id FROM usage_records WHERE request_id=?)`, requestID, if err != nil { return false, fmt.Errorf("关联请求终态索引: %w", err) } - linked, _ := result.RowsAffected() - if linked > 0 { + var linked bool + if err := tx.QueryRowContext(ctx, ` +SELECT EXISTS ( + SELECT 1 FROM request_detail_index + WHERE usage_id IN (SELECT id FROM usage_records WHERE request_id=?) +)`, requestID).Scan(&linked); err != nil { + return false, fmt.Errorf("确认请求终态索引关联: %w", err) + } + if linked { if _, err := tx.ExecContext(ctx, `UPDATE request_detail_index SET endpoint_kind=`+endpointKindSQL(`COALESCE(NULLIF((SELECT endpoint FROM usage_records WHERE id=request_detail_index.usage_id), ''), (SELECT endpoint FROM request_records WHERE request_id=?), '')`)+` WHERE usage_id IN (SELECT id FROM usage_records WHERE request_id=?)`, requestID, requestID, requestID, requestID, requestID); err != nil { return false, fmt.Errorf("更新请求端点索引: %w", err) } diff --git a/internal/repository/sqlite_usage.go b/internal/repository/sqlite_usage.go index 4e2139e..5145e78 100644 --- a/internal/repository/sqlite_usage.go +++ b/internal/repository/sqlite_usage.go @@ -12,6 +12,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "billing/internal/collection" @@ -231,10 +232,10 @@ ON billing_ledger(managed_key_id, id DESC); ` const ( - // CLIProxyAPI UsageRecord currently has no RequestID. The usage and terminal - // callbacks nevertheless carry request-start timestamps from the same host - // execution, normally only a few milliseconds apart. This window is used - // only for the read projection and never changes persisted billing facts. + // Older CLIProxyAPI schemas have no RequestID in UsageRecord. Their usage and + // terminal callbacks nevertheless carry request-start timestamps from the + // same host execution, normally only a few milliseconds apart. This window + // is used only for the read projection and never changes persisted billing facts. orphanLifecycleMatchWindow = 250 * time.Millisecond // Near-equal candidates are deliberately left separate instead of risking a // cross-request association under concurrent same-model traffic. @@ -243,8 +244,9 @@ const ( // SQLiteUsageRepository 使用单写连接保存事实，并用独立连接池承载只读查询。 type SQLiteUsageRepository struct { - db *sql.DB - readDB *sql.DB + db *sql.DB + readDB *sql.DB + projectionMu sync.Mutex } func OpenSQLiteUsage(databasePath string) (*SQLiteUsageRepository, error) { @@ -356,6 +358,8 @@ func sqliteDSN(databasePath string) (string, error) { } func (r *SQLiteUsageRepository) Insert(ctx context.Context, record collection.Record) error { + r.projectionMu.Lock() + defer r.projectionMu.Unlock() eventKey := usageBillingEventKey(record) tx, err := r.db.BeginTx(ctx, nil) if err != nil { @@ -468,6 +472,8 @@ UPDATE billing_cycles SET spent_micros=spent_micros+? WHERE managed_key_id=? AND } func (r *SQLiteUsageRepository) UpsertRequest(ctx context.Context, record collection.RequestRecord) error { + r.projectionMu.Lock() + defer r.projectionMu.Unlock() requestID := strings.TrimSpace(record.RequestID) if requestID == "" { return errors.New("request_id 不能为空") diff --git a/internal/repository/sqlite_usage_test.go b/internal/repository/sqlite_usage_test.go index 8d71515..9dab06a 100644 --- a/internal/repository/sqlite_usage_test.go +++ b/internal/repository/sqlite_usage_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "path/filepath" + "sync" "testing" "time" @@ -162,6 +163,126 @@ func TestSQLiteUsageMergesOrphanUsageWithLifecycleProjection(t *testing.T) { } } +func TestSQLiteUsageMergesExactRequestIdentityInEitherOrder(t *testing.T) { + for _, order := range []string{"usage-first", "lifecycle-first"} { + t.Run(order, func(t *testing.T) { + store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + startedAt := time.Date(2026, 8, 15, 18, 31, 43, 160_000_000, time.UTC) + usage := collection.Record{ + ManagedKeyID: "key-exact", RequestID: "request-exact", TraceID: "trace-exact", + RequestedAt: startedAt.Add(10 * time.Millisecond), APIKey: "exact-000000", + Model: "deepseek-v4-flash", ExecutorType: "CodexExecutor", RequestType: "SSE", + InputTokens: 90, OutputTokens: 15, TotalTokens: 105, + } + lifecycle := collection.RequestRecord{ + ManagedKeyID: "key-exact", RequestID: "request-exact", TraceID: "trace-exact", + RequestedAt: startedAt, CompletedAt: startedAt.Add(time.Second), + Model: "deepseek-v4-flash", SourceFormat: "openai-response", Stream: true, + Outcome: "succeeded", StatusCode: 200, Endpoint: "/v1/responses", + } + + insertUsage := func() { + if err := store.Insert(context.Background(), usage); err != nil { + t.Fatal(err) + } + } + insertLifecycle := func() { + if err := store.UpsertRequest(context.Background(), lifecycle); err != nil { + t.Fatal(err) + } + } + if order == "usage-first" { + insertUsage() + insertLifecycle() + } else { + insertLifecycle() + insertUsage() + } + + records, err := store.ListRecent(context.Background(), 10) + if err != nil { + t.Fatal(err) + } + if len(records) != 1 { + t.Fatalf("records = %d, want one exact request: %+v", len(records), records) + } + got := records[0] + if got.RequestID != "request-exact" || got.TraceID != "trace-exact" || got.Outcome != "succeeded" || + got.StatusCode != 200 || got.TotalTokens != 105 { + t.Fatalf("merged exact record = %+v", got) + } + page, err := store.QueryUsage(context.Background(), collection.UsageQuery{RequestID: "request-exact", PageSize: 10}) + if err != nil { + t.Fatal(err) + } + if page.Total != 1 || len(page.Records) != 1 || page.Records[0].TotalTokens != 105 { + t.Fatalf("projected exact records = %+v", page) + } + }) + } +} + +func TestSQLiteUsageMergesExactRequestIdentityConcurrently(t *testing.T) { + store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + const requestCount = 64 + startedAt := time.Date(2026, 8, 15, 18, 40, 0, 0, time.UTC) + start := make(chan struct{}) + errors := make(chan error, requestCount*2) + var workers sync.WaitGroup + for index := 0; index < requestCount; index++ { + requestID := fmt.Sprintf("request-exact-%03d", index) + workers.Add(2) + go func() { + defer workers.Done() + <-start + errors <- store.Insert(context.Background(), collection.Record{ + ManagedKeyID: "key-exact", RequestID: requestID, TraceID: "trace-exact", + RequestedAt: startedAt, Model: "deepseek-v4-flash", TotalTokens: 105, + }) + }() + go func() { + defer workers.Done() + <-start + errors <- store.UpsertRequest(context.Background(), collection.RequestRecord{ + ManagedKeyID: "key-exact", RequestID: requestID, TraceID: "trace-exact", + RequestedAt: startedAt, CompletedAt: startedAt.Add(time.Second), + Model: "deepseek-v4-flash", Stream: true, Outcome: "succeeded", StatusCode: 200, + }) + }() + } + close(start) + workers.Wait() + close(errors) + for err := range errors { + if err != nil { + t.Fatal(err) + } + } + + page, err := store.QueryUsage(context.Background(), collection.UsageQuery{KeyID: "key-exact", PageSize: 100}) + if err != nil { + t.Fatal(err) + } + if page.Total != requestCount || len(page.Records) != requestCount { + t.Fatalf("projected records = %d/%d, want %d: %+v", len(page.Records), page.Total, requestCount, page.Records) + } + for _, record := range page.Records { + if record.RequestID == "" || record.TotalTokens != 105 || record.Outcome != "succeeded" || record.StatusCode != 200 { + t.Fatalf("incomplete concurrent merge = %+v", record) + } + } +} + func TestSQLiteUsageMergesResolvedModelWithManagedKeyAlias(t *testing.T) { store, err := repository.OpenSQLiteUsage(filepath.Join(t.TempDir(), "usage.db")) if err != nil { diff --git a/patch/README.md b/patch/README.md new file mode 100644 index 0000000..f531419 --- /dev/null +++ b/patch/README.md @@ -0,0 +1,132 @@ +# CLIProxyAPI 宿主补丁 + +这里保存 billing 在 CLIProxyAPI `v7.2.132` 上需要的三项宿主修复。三项补丁都不改变计费规则或数据库结构，必须按下列顺序应用： + +1. [`cli-proxy-api-usage-context.patch`](cli-proxy-api-usage-context.patch) +2. [`cli-proxy-api-usage-identity.patch`](cli-proxy-api-usage-identity.patch) +3. [`cli-proxy-api-request-lifecycle-cancel.patch`](cli-proxy-api-request-lifecycle-cancel.patch) + +适用基线： + +- CLIProxyAPI：`v7.2.132` +- 提交：`78f0c4079e3e6273d65d03b5549cffc898703264` +- Native ABI：`1` +- 第二项补丁把 RPC schema 从 `3` 提升到 `4` + +升级 CLIProxyAPI 后不要直接假设补丁仍然需要或仍能应用。应先在新版本复测，再根据新版本的宿主契约重新核对。 + +## 1. 已取消上下文导致动态插件漏记 + +使用 `/v1/responses` 且请求体为 `"stream": false` 时，上游请求能够成功，CLIProxyAPI 内置用量统计也会增加，但动态用量插件可能完全收不到 `usage.handle`： + +- billing 没有新增用量事实； +- 用户余额没有扣减； +- 内置统计和动态插件统计不一致。 + +CLIProxyAPI 会异步分发最终用量记录，但分发任务沿用了原始 HTTP 请求上下文。非流式响应很快结束后，该上下文会被取消；动态插件加载器在调用 C ABI 前检查 `ctx.Done()`，于是跳过插件调用。内置插件不经过动态 C ABI 检查，所以仍能收到记录。 + +第一项补丁在动态用量适配器入口使用 `context.WithoutCancel`： + +- 保留上下文值； +- 不再继承已经结束的 HTTP 请求的取消信号和截止时间； +- 每个最终用量仍只调用一次插件，不增加逐块回调； +- `nil` 上下文退回 `context.Background()`。 + +## 2. Usage 缺少请求身份导致并发拆行 + +CLIProxyAPI 的请求生命周期会生成唯一 Request ID，并通过 `request.complete` 发送给插件，但 schema v3 的 `usage.handle` 没有携带该身份。billing 只能按用户、模型和时间窗口关联两类事实。 + +串行请求通常可以唯一匹配；同用户、同模型高并发时会出现多个等价候选。billing 为避免错配，保守地保留一条 request-only 和一条 usage-only，因此请求明细数量可能翻倍，但底层用量和扣费并未重复。 + +第二项补丁引入 RPC schema v4： + +- 请求生命周期创建后，把 Request ID 和父 Trace ID 写入实际执行上下文； +- `usage.handle` 增加可选的 `RequestID` 与 `TraceID`； +- 只有协商到 schema v4 的插件收到新增字段； +- schema v1–v3 插件保持原有载荷，能够继续加载； +- billing v4 按 Request ID 精确关联，旧宿主和历史数据仍走保守兼容逻辑。 + +补丁不虚构 Execution ID。一次生命周期内可能存在多个上游尝试时，只有宿主能够提供真实的逐尝试身份；当前修复只传递已经存在且语义明确的生命周期 Request ID。 + +## 3. SSE 启动阶段取消可能缺少请求终态 + +billing 在请求拦截阶段创建并发占用，依赖 CPA 后续发送唯一的 `request.complete` 释放占用。原实现只在同步执行启动完成后，由流消费协程观察执行上下文的取消并发送终态。 + +如果执行上下文恰好在上游连接、鉴权选择或首段流初始化期间取消，流消费协程可能尚未建立。此时 `request.complete` 必须等待同步启动调用返回；若自定义执行器或异常上游没有及时响应取消，就会表现为并发槽位长期占用。内置 Codex 执行器通常会随 context 立即返回，所以这是生命周期层的防御性加固，不是内置执行器的稳定必现缺陷。 + +第三项补丁在请求生命周期创建时直接注册上下文取消回调： + +- 不依赖执行器或流转发协程是否已经启动； +- 取消后立即发送 `outcome=canceled`、`status_code=0` 的终态； +- 正常完成时主动撤销取消回调； +- 与原有完成路径共用 `sync.Once`，两条路径竞态时仍只发送一个 `request.complete`； +- 不增加轮询、定时器或逐流块处理。 + +该补丁只能处理已经到达 CPA 执行上下文的取消信号。Windows 客户端通过 WSL localhost 转发访问 WSL 内 CPA 时，客户端断开可能不会立即传递到后端连接；这种情况下 CPA 看不到取消，任何生命周期回调都无法提前触发。生产 Linux 直连和 WSL 内直连不经过该转发层。 + +## 应用 + +在干净的 CLIProxyAPI 工作区执行： + +```bash +git checkout 78f0c4079e3e6273d65d03b5549cffc898703264 +git apply --check /path/to/billing/patch/cli-proxy-api-usage-context.patch +git apply /path/to/billing/patch/cli-proxy-api-usage-context.patch +git apply --check /path/to/billing/patch/cli-proxy-api-usage-identity.patch +git apply /path/to/billing/patch/cli-proxy-api-usage-identity.patch +git apply --check /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch +git apply /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch +gofmt -w internal/pluginhost sdk/api/handlers sdk/cliproxy/usage sdk/pluginabi sdk/pluginapi +go test -race ./internal/pluginhost ./sdk/api/handlers ./sdk/cliproxy/usage -count=1 +CGO_ENABLED=1 go build -o test-output ./cmd/server +rm test-output +``` + +billing 必须使用 RPC schema v4 重新构建；旧 billing 动态库会向下协商到旧 schema，无法获得 Usage 请求身份。 + +## 验证标准 + +运行时至少交叉核对： + +1. 分别发送一次非流式和一次 SSE 请求； +2. 两次请求均只产生一条明细和一次扣费； +3. 并发发送相同用户、相同模型的混合请求； +4. N 个 HTTP 200 最终对应 N 条明细、N 个唯一 Request ID、N 条底层 Usage 和 N 次扣费； +5. CLIProxyAPI 内置队列、billing 和独立 keeper 的增量一致； +6. 等待异步任务稳定后没有迟到重复； +7. 请求正常结束后并发占用回到零； +8. 在上游启动阶段和已开始传输后分别强制中断 SSE，两次都只产生一个 canceled 终态且并发占用回到零； +9. 日志中没有插件错误或 panic。 + +## 已验证结果 + +三个补丁与 billing schema v4 使用真实 DeepSeek 上游完成了本地验证： + +- `cpads` 有 Key 成功、无 Key 返回 401； +- 非流式和 SSE 各一次：2 条明细、2 次扣费，等待 4 秒无重复； +- 12 个混合并发请求：12 次 HTTP 200、12 条合并明细、12 个唯一 Request ID、12 次扣费； +- 流式与非流式各 6 条，终态、状态码、Token 和 Trace ID 完整； +- keeper 同步增加 12 条，等待 5 秒没有迟到重复； +- WSL 内直连分别在 50ms（首字节前）和 1s（流传输中）强制断开：两次均只有一条 canceled 明细，并发占用立即回到零； +- 仅前两项补丁与三项补丁在内置 Codex 执行器上的直连取消 A/B 都能正常终止；第三项补丁额外覆盖执行器启动阻塞且未响应 context 的单元场景； +- 请求正常结束后 `active_requests=0`，空闲 5 秒没有持续 CPU 占用； +- billing 全部测试、billing race 测试及 CLIProxyAPI 相关包 race 测试通过； +- schema v3 对照插件仍能正常注册，新增字段不会发送给旧 schema。 + +CLIProxyAPI 全仓 `go test ./...` 在该基线自身仍有 5 个无关失败：3 个 Claude OS 指纹用例在 WSL 中得到 `MacOS` 而期望 `Linux`，以及 2 个 `reviewedInPlaceByteWrites` 清单陈旧项。补丁涉及的包及其 race 测试均通过。 + +## 性能影响 + +三项修复都不增加队列、重试或流式分块回调。每个最终用量和请求终态仍最多调用插件一次；新增工作是创建轻量上下文包装、在 schema v4 JSON 中增加两个短字符串，以及为活跃请求注册一个随正常完成立即撤销的标准库取消回调。并发验证中没有观察到持续 CPU 或内存增长。 + +## 回滚 + +源码必须按应用顺序的反方向回滚： + +```bash +git apply -R /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch +git apply -R /path/to/billing/patch/cli-proxy-api-usage-identity.patch +git apply -R /path/to/billing/patch/cli-proxy-api-usage-context.patch +``` + +部署回滚应恢复升级前保存的 CLIProxyAPI 和 billing 动态库后重启宿主。数据库不需要回滚；schema v4 只让新记录获得精确身份，不改变历史账目。 diff --git a/patch/cli-proxy-api-request-lifecycle-cancel.patch b/patch/cli-proxy-api-request-lifecycle-cancel.patch new file mode 100644 index 0000000..69cb568 --- /dev/null +++ b/patch/cli-proxy-api-request-lifecycle-cancel.patch @@ -0,0 +1,160 @@ +diff --git a/sdk/api/handlers/handlers_interceptors.go b/sdk/api/handlers/handlers_interceptors.go +--- a/sdk/api/handlers/handlers_interceptors.go ++++ b/sdk/api/handlers/handlers_interceptors.go +@@ -1,6 +1,7 @@ + package handlers + + import ( ++ stdcontext "context" + "net/http" + "sync" + "time" +@@ -66,6 +67,9 @@ type requestLifecycleSkipHost interface { + + type requestLifecycleTracker struct { + once sync.Once ++ cancelMu sync.Mutex ++ cancelStop func() bool ++ completed bool + ctx context.Context + host PluginInterceptorHost + skipPluginID string +@@ -76,7 +80,7 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF + requestID := uuid.NewString() + traceID := logging.GetRequestID(ctx) + ctx = coreusage.WithRequestIdentity(ctx, requestID, traceID) +- return &requestLifecycleTracker{ ++ tracker := &requestLifecycleTracker{ + ctx: ctx, + host: h.interceptorHost(), + skipPluginID: skipPluginID, +@@ -91,6 +95,39 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF + Metadata: metadata, + }, + } ++ tracker.watchCancellation(ctx) ++ return tracker ++} ++ ++func (t *requestLifecycleTracker) watchCancellation(ctx context.Context) { ++ if t == nil || ctx == nil || ctx.Done() == nil { ++ return ++ } ++ stop := stdcontext.AfterFunc(ctx, func() { ++ t.complete(pluginapi.RequestCompletionCanceled, 0, ctx.Err()) ++ }) ++ t.cancelMu.Lock() ++ if t.completed { ++ t.cancelMu.Unlock() ++ stop() ++ return ++ } ++ t.cancelStop = stop ++ t.cancelMu.Unlock() ++} ++ ++func (t *requestLifecycleTracker) stopCancellationWatch() { ++ if t == nil { ++ return ++ } ++ t.cancelMu.Lock() ++ t.completed = true ++ stop := t.cancelStop ++ t.cancelStop = nil ++ t.cancelMu.Unlock() ++ if stop != nil { ++ stop() ++ } + } + + func (t *requestLifecycleTracker) executionContext() context.Context { +@@ -112,6 +149,7 @@ func (t *requestLifecycleTracker) complete(outcome pluginapi.RequestCompletionOu + return + } + t.once.Do(func() { ++ t.stopCancellationWatch() + completion := t.completion + completion.Outcome = outcome + completion.StatusCode = statusCode +diff --git a/sdk/api/handlers/handlers_interceptors_test.go b/sdk/api/handlers/handlers_interceptors_test.go +--- a/sdk/api/handlers/handlers_interceptors_test.go ++++ b/sdk/api/handlers/handlers_interceptors_test.go +@@ -516,6 +516,78 @@ func TestHandlerLifecycleCompletesCanceledStream(t *testing.T) { + } + } + ++func TestHandlerLifecycleCompletesCanceledStreamWhileExecutorIsStarting(t *testing.T) { ++ model := "handler-interceptor-lifecycle-canceled-before-stream-start" ++ executorStarted := make(chan struct{}) ++ releaseExecutor := make(chan struct{}) ++ executor := &interceptorCaptureExecutor{ ++ stream: func(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (*coreexecutor.StreamResult, error) { ++ close(executorStarted) ++ <-releaseExecutor ++ chunks := make(chan coreexecutor.StreamChunk) ++ close(chunks) ++ return &coreexecutor.StreamResult{Chunks: chunks}, nil ++ }, ++ } ++ handler := newInterceptorHandler(t, model, executor, &sdkconfig.SDKConfig{}) ++ completions := make(chan pluginapi.RequestCompletion, 2) ++ handler.SetPluginHost(&handlerInterceptorTestHost{ ++ completeRequest: func(_ context.Context, completion pluginapi.RequestCompletion) { ++ completions <- completion ++ }, ++ }) ++ ++ ctx, cancel := context.WithCancel(context.Background()) ++ type streamResult struct { ++ data <-chan []byte ++ errs <-chan *interfaces.ErrorMessage ++ } ++ result := make(chan streamResult, 1) ++ go func() { ++ data, _, errs := handler.ExecuteStreamWithAuthManager(ctx, "openai", model, []byte(`{"model":"`+model+`","stream":true}`), "") ++ result <- streamResult{data: data, errs: errs} ++ }() ++ ++ select { ++ case <-executorStarted: ++ case <-time.After(time.Second): ++ close(releaseExecutor) ++ t.Fatal("executor did not start") ++ } ++ cancel() ++ ++ select { ++ case completion := <-completions: ++ if completion.Outcome != pluginapi.RequestCompletionCanceled || completion.StatusCode != 0 || completion.Error == "" { ++ close(releaseExecutor) ++ t.Fatalf("completion = %#v", completion) ++ } ++ case <-time.After(time.Second): ++ close(releaseExecutor) ++ t.Fatal("missing cancellation completion while executor startup was blocked") ++ } ++ ++ close(releaseExecutor) ++ channels := <-result ++ for channels.data != nil || channels.errs != nil { ++ select { ++ case _, ok := <-channels.data: ++ if !ok { ++ channels.data = nil ++ } ++ case _, ok := <-channels.errs: ++ if !ok { ++ channels.errs = nil ++ } ++ } ++ } ++ select { ++ case duplicate := <-completions: ++ t.Fatalf("duplicate stream completion = %#v", duplicate) ++ default: ++ } ++} ++ + func TestHandlerRequestInterceptorRewritesExecutorRequest(t *testing.T) { + model := "handler-interceptor-request-model" + executor := &interceptorCaptureExecutor{} diff --git a/patch/cli-proxy-api-usage-context.patch b/patch/cli-proxy-api-usage-context.patch new file mode 100644 index 0000000..d849b4a --- /dev/null +++ b/patch/cli-proxy-api-usage-context.patch @@ -0,0 +1,60 @@ +diff --git a/internal/pluginhost/adapters_test.go b/internal/pluginhost/adapters_test.go +index de62918e..302b1f38 100644 +--- a/internal/pluginhost/adapters_test.go ++++ b/internal/pluginhost/adapters_test.go +@@ -2234,6 +2234,39 @@ func TestUsageAdapterPanicFusesPlugin(t *testing.T) { + } + } + ++func TestUsageAdapterDetachesCanceledRequestContext(t *testing.T) { ++ type contextKey struct{} ++ const contextValue = "usage-context-value" ++ ++ called := 0 ++ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) { ++ called++ ++ if errContext := ctx.Err(); errContext != nil { ++ t.Fatalf("usage context error = %v, want nil", errContext) ++ } ++ if value := ctx.Value(contextKey{}); value != contextValue { ++ t.Fatalf("usage context value = %v, want %q", value, contextValue) ++ } ++ }) ++ host := newHostWithRecords(capabilityRecord{ ++ id: "usage-canceled-context", ++ plugin: pluginapi.Plugin{Capabilities: pluginapi.Capabilities{ ++ UsagePlugin: plugin, ++ }}, ++ }) ++ adapter := &usageAdapter{ ++ host: host, ++ pluginID: "usage-canceled-context", ++ } ++ ++ ctx, cancel := context.WithCancel(context.WithValue(context.Background(), contextKey{}, contextValue)) ++ cancel() ++ adapter.HandleUsage(ctx, coreusage.Record{Provider: "plugin-provider"}) ++ if called != 1 { ++ t.Fatalf("usage plugin calls = %d, want 1", called) ++ } ++} ++ + func TestUsageAdapterNormalizesOmittedGenerateToTrue(t *testing.T) { + var gotGenerate bool + plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) { +diff --git a/internal/pluginhost/adapters_usage_translation.go b/internal/pluginhost/adapters_usage_translation.go +index 2201eb6c..93a4c8a5 100644 +--- a/internal/pluginhost/adapters_usage_translation.go ++++ b/internal/pluginhost/adapters_usage_translation.go +@@ -132,6 +132,11 @@ func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record) + if a == nil { + return + } ++ if ctx == nil { ++ ctx = context.Background() ++ } else { ++ ctx = context.WithoutCancel(ctx) ++ } + plugin := a.host.currentUsagePlugin(a.pluginID) + if plugin == nil { + return diff --git a/patch/cli-proxy-api-usage-identity.patch b/patch/cli-proxy-api-usage-identity.patch new file mode 100644 index 0000000..fb43897 --- /dev/null +++ b/patch/cli-proxy-api-usage-identity.patch @@ -0,0 +1,348 @@ +diff --git a/internal/pluginhost/adapters_test.go b/internal/pluginhost/adapters_test.go +index 302b1f38..610a3753 100644 +--- a/internal/pluginhost/adapters_test.go ++++ b/internal/pluginhost/adapters_test.go +@@ -2266,6 +2266,54 @@ func TestUsageAdapterDetachesCanceledRequestContext(t *testing.T) { + } + } + ++func TestUsageAdapterIncludesRequestIdentityForSchemaFour(t *testing.T) { ++ var got pluginapi.UsageRecord ++ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) { ++ got = record ++ }) ++ host := newHostWithRecords(capabilityRecord{ ++ id: "usage-identity", ++ plugin: pluginapi.Plugin{SchemaVersion: pluginabi.SchemaVersionUsageIdentity, Capabilities: pluginapi.Capabilities{ ++ UsagePlugin: plugin, ++ }}, ++ }) ++ adapter := &usageAdapter{ ++ host: host, ++ pluginID: "usage-identity", ++ schemaVersion: pluginabi.SchemaVersionUsageIdentity, ++ } ++ ++ ctx := coreusage.WithRequestIdentity(context.Background(), "request-1", "trace-1") ++ adapter.HandleUsage(ctx, coreusage.Record{Provider: "provider"}) ++ if got.RequestID != "request-1" || got.TraceID != "trace-1" { ++ t.Fatalf("usage identity = %q, %q", got.RequestID, got.TraceID) ++ } ++} ++ ++func TestUsageAdapterOmitsRequestIdentityForLegacySchema(t *testing.T) { ++ var got pluginapi.UsageRecord ++ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) { ++ got = record ++ }) ++ host := newHostWithRecords(capabilityRecord{ ++ id: "usage-identity-legacy", ++ plugin: pluginapi.Plugin{SchemaVersion: pluginabi.SchemaVersionUsageIdentity - 1, Capabilities: pluginapi.Capabilities{ ++ UsagePlugin: plugin, ++ }}, ++ }) ++ adapter := &usageAdapter{ ++ host: host, ++ pluginID: "usage-identity-legacy", ++ schemaVersion: pluginabi.SchemaVersionUsageIdentity - 1, ++ } ++ ++ ctx := coreusage.WithRequestIdentity(context.Background(), "request-1", "trace-1") ++ adapter.HandleUsage(ctx, coreusage.Record{Provider: "provider"}) ++ if got.RequestID != "" || got.TraceID != "" { ++ t.Fatalf("legacy usage identity = %q, %q", got.RequestID, got.TraceID) ++ } ++} ++ + func TestUsageAdapterNormalizesOmittedGenerateToTrue(t *testing.T) { + var gotGenerate bool + plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) { +diff --git a/internal/pluginhost/adapters_usage_translation.go b/internal/pluginhost/adapters_usage_translation.go +index 93a4c8a5..ad046780 100644 +--- a/internal/pluginhost/adapters_usage_translation.go ++++ b/internal/pluginhost/adapters_usage_translation.go +@@ -10,6 +10,7 @@ import ( + "github.com/router-for-me/CLIProxyAPI/v7/internal/registry" + "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" + coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage" ++ "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginabi" + "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" + log "github.com/sirupsen/logrus" +@@ -26,9 +27,10 @@ func (h *Host) RegisterUsagePlugins() { + continue + } + coreusage.RegisterNamedPlugin("plugin:"+record.id, &usageAdapter{ +- host: h, +- pluginID: record.id, +- plugin: plugin, ++ host: h, ++ pluginID: record.id, ++ plugin: plugin, ++ schemaVersion: record.plugin.SchemaVersion, + }) + } + } +@@ -114,9 +116,10 @@ func (h *Host) isPluginFused(id string) bool { + } + + type usageAdapter struct { +- host *Host +- pluginID string +- plugin pluginapi.UsagePlugin ++ host *Host ++ pluginID string ++ plugin pluginapi.UsagePlugin ++ schemaVersion uint32 + } + + type thinkingAdapter struct { +@@ -146,7 +149,13 @@ func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record) + a.host.fusePlugin(a.pluginID, "UsagePlugin.HandleUsage", recovered) + } + }() ++ requestID, traceID := "", "" ++ if a.schemaVersion >= pluginabi.SchemaVersionUsageIdentity { ++ requestID, traceID = coreusage.RequestIdentityFromContext(ctx) ++ } + plugin.HandleUsage(ctx, pluginapi.UsageRecord{ ++ RequestID: requestID, ++ TraceID: traceID, + Provider: record.Provider, + ExecutorType: record.ExecutorType, + Model: record.Model, +diff --git a/sdk/api/handlers/handlers_execution.go b/sdk/api/handlers/handlers_execution.go +index 994bd33a..3cc7743e 100644 +--- a/sdk/api/handlers/handlers_execution.go ++++ b/sdk/api/handlers/handlers_execution.go +@@ -71,6 +71,7 @@ func (h *BaseAPIHandler) executeWithAuthManagerFormats(ctx context.Context, entr + } + afterAuthCapture := &requestAfterAuthCapture{} + lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, normalizedModel, originalRequestedModel, false, reqMeta, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + opts := coreexecutor.Options{ + Stream: false, + Alt: alt, +@@ -136,6 +137,7 @@ func (h *BaseAPIHandler) executeCountWithAuthManager(ctx context.Context, handle + } + afterAuthCapture := &requestAfterAuthCapture{} + lifecycle := h.newRequestLifecycleTracker(ctx, handlerType, normalizedModel, originalRequestedModel, false, reqMeta, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + opts := coreexecutor.Options{ + Stream: false, + Alt: alt, +@@ -177,6 +179,7 @@ func (h *BaseAPIHandler) executeWithPluginExecutor(ctx context.Context, entryPro + } + req, opts := h.pluginExecutorRequest(ctx, entryProtocol, responseProtocol, modelName, originalRequestedModel, rawJSON, alt, false, execOptions) + lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, modelName, originalRequestedModel, false, opts.Metadata, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + var interceptErr *interfaces.ErrorMessage + req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, entryProtocol, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID) + if interceptErr != nil { +@@ -211,6 +214,7 @@ func (h *BaseAPIHandler) countWithPluginExecutor(ctx context.Context, handlerTyp + } + req, opts := h.pluginExecutorRequest(ctx, handlerType, handlerType, modelName, originalRequestedModel, rawJSON, alt, false, execOptions) + lifecycle := h.newRequestLifecycleTracker(ctx, handlerType, modelName, originalRequestedModel, false, opts.Metadata, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + var interceptErr *interfaces.ErrorMessage + req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, handlerType, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID) + if interceptErr != nil { +diff --git a/sdk/api/handlers/handlers_interceptors.go b/sdk/api/handlers/handlers_interceptors.go +index a8b35604..d27198c9 100644 +--- a/sdk/api/handlers/handlers_interceptors.go ++++ b/sdk/api/handlers/handlers_interceptors.go +@@ -9,6 +9,7 @@ import ( + "github.com/router-for-me/CLIProxyAPI/v7/internal/interfaces" + "github.com/router-for-me/CLIProxyAPI/v7/internal/logging" + coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" ++ coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage" + "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi" + "golang.org/x/net/context" + ) +@@ -74,6 +75,7 @@ type requestLifecycleTracker struct { + func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceFormat, model, requestedModel string, stream bool, metadata map[string]any, skipPluginID string) *requestLifecycleTracker { + requestID := uuid.NewString() + traceID := logging.GetRequestID(ctx) ++ ctx = coreusage.WithRequestIdentity(ctx, requestID, traceID) + return &requestLifecycleTracker{ + ctx: ctx, + host: h.interceptorHost(), +@@ -91,6 +93,13 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF + } + } + ++func (t *requestLifecycleTracker) executionContext() context.Context { ++ if t == nil || t.ctx == nil { ++ return context.Background() ++ } ++ return t.ctx ++} ++ + func (t *requestLifecycleTracker) requestID() string { + if t == nil { + return "" +diff --git a/sdk/api/handlers/handlers_interceptors_test.go b/sdk/api/handlers/handlers_interceptors_test.go +index 0b328f3e..1ad85527 100644 +--- a/sdk/api/handlers/handlers_interceptors_test.go ++++ b/sdk/api/handlers/handlers_interceptors_test.go +@@ -16,6 +16,7 @@ import ( + "github.com/router-for-me/CLIProxyAPI/v7/internal/registry" + coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" + coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" ++ coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage" + sdkconfig "github.com/router-for-me/CLIProxyAPI/v7/sdk/config" + "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi" + ) +@@ -230,6 +231,11 @@ func TestRequestLifecycleTrackerUsesUniqueExecutionIDs(t *testing.T) { + if first.completion.TraceID != "trace-1" || second.completion.TraceID != "trace-1" { + t.Fatalf("trace IDs = %q and %q", first.completion.TraceID, second.completion.TraceID) + } ++ firstRequestID, firstTraceID := coreusage.RequestIdentityFromContext(first.executionContext()) ++ secondRequestID, secondTraceID := coreusage.RequestIdentityFromContext(second.executionContext()) ++ if firstRequestID != first.requestID() || secondRequestID != second.requestID() || firstTraceID != "trace-1" || secondTraceID != "trace-1" { ++ t.Fatalf("usage identities = (%q, %q) and (%q, %q)", firstRequestID, firstTraceID, secondRequestID, secondTraceID) ++ } + } + + func TestHandlerRequestInterceptorTerminatesBeforeAuth(t *testing.T) { +diff --git a/sdk/api/handlers/handlers_stream.go b/sdk/api/handlers/handlers_stream.go +index 9cceec54..dcdc08df 100644 +--- a/sdk/api/handlers/handlers_stream.go ++++ b/sdk/api/handlers/handlers_stream.go +@@ -41,6 +41,7 @@ func (h *BaseAPIHandler) streamWithPluginExecutor(ctx context.Context, entryProt + } + req, opts := h.pluginExecutorRequest(ctx, entryProtocol, responseProtocol, modelName, originalRequestedModel, rawJSON, alt, true, execOptions) + lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, modelName, originalRequestedModel, true, opts.Metadata, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + var interceptErr *interfaces.ErrorMessage + req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, entryProtocol, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID) + if interceptErr != nil { +@@ -305,6 +306,7 @@ func (h *BaseAPIHandler) executeStreamWithAuthManagerFormats(ctx context.Context + } + afterAuthCapture := &requestAfterAuthCapture{} + lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, normalizedModel, originalRequestedModel, true, reqMeta, execOptions.SkipInterceptorPluginID) ++ ctx = lifecycle.executionContext() + opts := coreexecutor.Options{ + Stream: true, + Alt: alt, +diff --git a/sdk/cliproxy/usage/manager.go b/sdk/cliproxy/usage/manager.go +index ca36dc55..13506fbb 100644 +--- a/sdk/cliproxy/usage/manager.go ++++ b/sdk/cliproxy/usage/manager.go +@@ -78,6 +78,34 @@ type requestedModelAliasContextKey struct{} + type reasoningEffortContextKey struct{} + type serviceTierContextKey struct{} + type generateContextKey struct{} ++type requestIdentityContextKey struct{} ++ ++type requestIdentity struct { ++ requestID string ++ traceID string ++} ++ ++// WithRequestIdentity stores the model execution and parent request IDs for usage sinks. ++func WithRequestIdentity(ctx context.Context, requestID, traceID string) context.Context { ++ if ctx == nil { ++ ctx = context.Background() ++ } ++ requestID = strings.TrimSpace(requestID) ++ traceID = strings.TrimSpace(traceID) ++ if requestID == "" && traceID == "" { ++ return ctx ++ } ++ return context.WithValue(ctx, requestIdentityContextKey{}, requestIdentity{requestID: requestID, traceID: traceID}) ++} ++ ++// RequestIdentityFromContext returns the model execution and parent request IDs stored in ctx. ++func RequestIdentityFromContext(ctx context.Context) (requestID, traceID string) { ++ if ctx == nil { ++ return "", "" ++ } ++ identity, _ := ctx.Value(requestIdentityContextKey{}).(requestIdentity) ++ return strings.TrimSpace(identity.requestID), strings.TrimSpace(identity.traceID) ++} + + // WithRequestedModelAlias stores the client-requested model name for usage sinks. + func WithRequestedModelAlias(ctx context.Context, alias string) context.Context { +diff --git a/sdk/cliproxy/usage/manager_test.go b/sdk/cliproxy/usage/manager_test.go +index 6f7b1fbb..fc2b3120 100644 +--- a/sdk/cliproxy/usage/manager_test.go ++++ b/sdk/cliproxy/usage/manager_test.go +@@ -36,6 +36,21 @@ func TestGenerateFromContextHonorsExplicitFalse(t *testing.T) { + } + } + ++func TestRequestIdentityContextRoundTrip(t *testing.T) { ++ ctx := WithRequestIdentity(context.Background(), " request-1 ", " trace-1 ") ++ requestID, traceID := RequestIdentityFromContext(ctx) ++ if requestID != "request-1" || traceID != "trace-1" { ++ t.Fatalf("request identity = %q, %q", requestID, traceID) ++ } ++} ++ ++func TestRequestIdentityContextDefaultsEmpty(t *testing.T) { ++ requestID, traceID := RequestIdentityFromContext(nil) ++ if requestID != "" || traceID != "" { ++ t.Fatalf("empty request identity = %q, %q", requestID, traceID) ++ } ++} ++ + func TestRecordOmittedGenerateIsEnabled(t *testing.T) { + // Existing callers construct Record without setting Generate. + // Omission must remain distinguishable from explicit false and default to true. +diff --git a/sdk/pluginabi/types.go b/sdk/pluginabi/types.go +index 97c41a13..a8a3e470 100644 +--- a/sdk/pluginabi/types.go ++++ b/sdk/pluginabi/types.go +@@ -10,10 +10,13 @@ const ( + // Version 3 omits OriginalRequest/RequestBody on payload stream chunks + // (ChunkIndex >= 0); those fields remain on StreamChunkHeaderInitIndex only. + // Plugins that still need per-chunk request bodies should keep schema_version < 3. +- SchemaVersion uint32 = 3 ++ SchemaVersion uint32 = 4 + // SchemaVersionStreamChunkOmitRequestBody is the first schema version that omits + // request bodies on payload stream-chunk interceptor calls. + SchemaVersionStreamChunkOmitRequestBody uint32 = 3 ++ // SchemaVersionUsageIdentity is the first schema version that attaches ++ // request lifecycle identity to usage records. ++ SchemaVersionUsageIdentity uint32 = 4 + ) + + const ( +diff --git a/sdk/pluginabi/types_test.go b/sdk/pluginabi/types_test.go +index 8fa63542..86253848 100644 +--- a/sdk/pluginabi/types_test.go ++++ b/sdk/pluginabi/types_test.go +@@ -27,12 +27,15 @@ func TestEnvelopeRoundTrip(t *testing.T) { + } + + func TestMethodNamesAreStable(t *testing.T) { +- if SchemaVersion != 3 { +- t.Fatalf("SchemaVersion = %d, want 3", SchemaVersion) ++ if SchemaVersion != 4 { ++ t.Fatalf("SchemaVersion = %d, want 4", SchemaVersion) + } + if SchemaVersionStreamChunkOmitRequestBody != 3 { + t.Fatalf("SchemaVersionStreamChunkOmitRequestBody = %d, want 3", SchemaVersionStreamChunkOmitRequestBody) + } ++ if SchemaVersionUsageIdentity != 4 { ++ t.Fatalf("SchemaVersionUsageIdentity = %d, want 4", SchemaVersionUsageIdentity) ++ } + if MethodPluginRegister != "plugin.register" { + t.Fatalf("MethodPluginRegister = %q", MethodPluginRegister) + } +diff --git a/sdk/pluginapi/types.go b/sdk/pluginapi/types.go +index 6add5d69..894e3b77 100644 +--- a/sdk/pluginapi/types.go ++++ b/sdk/pluginapi/types.go +@@ -1316,6 +1316,10 @@ type ManagementResponse struct { + + // UsageRecord describes request usage and billing metadata. + type UsageRecord struct { ++ // RequestID identifies the request lifecycle associated with this usage record. ++ RequestID string ++ // TraceID identifies the parent inbound HTTP request when available. ++ TraceID string + // Provider identifies the upstream provider. + Provider string + // ExecutorType identifies the executor implementation. diff --git a/scripts/check-env.ps1 b/scripts/check-env.ps1 index 749158b..2c2dd66 100644 --- a/scripts/check-env.ps1 +++ b/scripts/check-env.ps1 @@ -13,5 +13,5 @@ function Resolve-Tool([string]$Name) { $go = Resolve-Tool 'go' Write-Host "Go: $(& $go version)" -Write-Host 'CLIProxyAPI target: ABI 1 / RPC schema 3' +Write-Host 'CLIProxyAPI target: ABI 1 / RPC schema 4' Write-Warning '本项目默认在 WSL/Linux 构建；请优先运行 wsl bash ./scripts/build.sh。' diff --git a/scripts/check-env.sh b/scripts/check-env.sh index f0abccf..d0570c0 100644 --- a/scripts/check-env.sh +++ b/scripts/check-env.sh @@ -13,4 +13,4 @@ command -v gcc >/dev/null 2>&1 || { echo "Go bootstrap: $(go version)" echo "Go effective: $(go env GOVERSION)" echo "GCC: $(gcc --version | head -n 1)" -echo 'CLIProxyAPI target: ABI 1 / RPC schema 3' +echo 'CLIProxyAPI target: ABI 1 / RPC schema 4' diff --git a/scripts/package.sh b/scripts/package.sh index 6a10804..58d1a08 100755 --- a/scripts/package.sh +++ b/scripts/package.sh @@ -29,8 +29,8 @@ printf '%s\n' \ "version=$version" \ "target=linux/amd64" \ "native_abi=1" \ - "rpc_schema=3" \ - "cliproxyapi_revision=f43aad7637ad813745bf7d341acb5663617570c5" \ + "rpc_schema=4" \ + "cliproxyapi_revision=78f0c4079e3e6273d65d03b5549cffc898703264+usage-context+usage-identity+request-lifecycle-cancel" \ >"$package_dir/VERSION.txt" archive="$root/dist/$name.tar.gz" diff --git a/scripts/stop-test-host.sh b/scripts/stop-test-host.sh index 1168b5a..69f3ed1 100644 --- a/scripts/stop-test-host.sh +++ b/scripts/stop-test-host.sh @@ -16,6 +16,10 @@ if [[ -n "$pid" ]] && kill -0 "$pid" 2>/dev/null; then kill -0 "$pid" 2>/dev/null || break sleep 0.25 done + if kill -0 "$pid" 2>/dev/null; then + echo "CPA test host did not stop within 5 seconds (PID $pid)." >&2 + exit 1 + fi fi rm -f "$pid_file" echo 'CPA test host stopped.'