From 549b5398a0abeedd27f652471a6f8d6ae6dceb5a Mon Sep 17 00:00:00 2001 From: Mag1cFall Date: Wed, 11 Nov 2026 04:48:54 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=8C=E5=96=84=20Carbon=20=E9=95=BF?= =?UTF-8?q?=E5=9B=9E=E7=AD=94=E7=9A=84=E8=87=AA=E5=8A=A8=E7=BB=AD=E5=86=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 续写轮沿用首轮写入的 proof，Camoufox 后端沿用页面为首轮写入的 proof 直接发送 - 续写轮遇到可换账户的错误时换到其他账户，以同一续写 token 重发该轮，额度失败的账户进入冷却 - 换账户后转发、认证检查与用量明细跟随当前账户，被换下的账户立即释放 - 账户运行状态锁在请求期限内等到释放 - 协议、开发与日志文档说明续写轮的 proof、换账户与锁等待 --- docs/development.md | 2 +- docs/logging.md | 2 +- docs/protocol.md | 8 +- internal/aistudio/accounts.go | 5 +- internal/aistudio/build_embed.go | 2 +- internal/aistudio/client.go | 3 +- internal/aistudio/generate.go | 39 +++++++- internal/aistudio/imagen.go | 2 +- internal/aistudio/runtime_native.go | 21 +++-- internal/aistudio/service.go | 28 +++--- internal/aistudio/types.go | 2 + internal/aistudio/waa.go | 6 +- internal/app/runtime.go | 131 +++++++++++++++++++-------- internal/camoufoxnative/generate.go | 33 +++---- internal/camoufoxnative/protected.go | 6 +- 15 files changed, 198 insertions(+), 92 deletions(-) diff --git a/docs/development.md b/docs/development.md index 66a78909..c021070c 100644 --- a/docs/development.md +++ b/docs/development.md @@ -313,7 +313,7 @@ POST /api/control/start `runtime-state.json` 的 `model_access` value 为 `{state,checked_at,reason?}`，成功状态为 `verified`；`cooldowns` value 为 `{until,reason?}`；`resources` value 保存 kind、name、mime、size、purpose、created_at 与可选 video 元数据。Drive file、视频任务、视频产物和 Bidi 恢复 token 均保持创建账户粘性。视频任务额外保存公开 video object 的 model、seconds、size 与 UTC 创建时间，生成服务或进程重启后的轮询继续投影相同字段。 -活动请求锁保护跨进程账户租约，短事务锁保护 `runtime-state.json` 合并写回。短事务锁位于 `auth/.leases/<账户>.runtime.lock`，以 25ms 间隔等待，最多 2 秒；取得锁后重读磁盘状态，只修改目标字段，原子替换文件，再同步内存与资源索引。取租约时同样在锁内读取所选账户的文件，内容与上次同步或写入时相同则沿用内存状态。接收 context 的资源事务会在取消或 deadline 到达时提前返回。 +活动请求锁保护跨进程账户租约，短事务锁保护 `runtime-state.json` 合并写回。短事务锁位于 `auth/.leases/<账户>.runtime.lock`，以 25ms 间隔等到锁释放；取得锁后重读磁盘状态，只修改目标字段，原子替换文件，再同步内存与资源索引。取租约时同样在锁内读取所选账户的文件，内容与上次同步或写入时相同则沿用内存状态。接收 context 的资源事务会在取消或 deadline 到达时提前返回。 账户更新以 `account.json` 原子写入为持久提交点。`internal/app` 先准备 `pending`（尚未提交）的固定出口，关闭旧 Worker并锁定该账户的 Worker 配置，再调用 `AccountLease.SaveConfig`；保存成功后依次提交 Worker 配置与固定出口。准备、关闭或保存失败时丢弃 pending 更新；保存后的租约释放错误原样返回，已发布配置继续生效。 diff --git a/docs/logging.md b/docs/logging.md index 14ebdce3..69483d98 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -200,7 +200,7 @@ ERROR account@example.com WAA Worker 停止失败 | PID=18240 | 耗时=10.002s | `上游首事件到达` | 首事件等待后解出第一个语义事件 | `等待`、`事件`、`模型` | | `事件流停顿` | 已有语义事件，连续 15 秒没有下一事件 | `最近事件`、`推理`、`正文`、`网络字节`、`最近网络` | | `事件流恢复` | 停顿后解出下一语义事件 | `停顿`、`当前事件` | -| `账号切换` | 当前账户在首个上游语义事件前失败，或非流式请求的上游流中途中断 | 当前账户来源、模型和下一行原始原因 | +| `账号切换` | 当前账户在首个上游语义事件前失败，或非流式请求的上游流中途中断；`账号切换 \| 续写轮` 为自动续写轮在当前账户上失败后换账户重发 | 当前账户来源、模型和下一行原始原因 | | `备用请求发起` | 原请求等待首个语义事件达到 `FIRST_EVENT_HEDGE` 的阈值，空闲账户开始同一请求 | `账户`、`原请求已等待`、`阈值`、`模型` | | `备用请求胜出` | 备用请求先解出首个语义事件，原请求取消 | `账户`、`原请求已等待`、`模型` | | `原请求胜出` | 原请求先解出首个语义事件，备用请求取消 | `备用账户`、`模型` | diff --git a/docs/protocol.md b/docs/protocol.md index f9627e6d..ea6012cc 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -183,7 +183,7 @@ Google 邮箱的小写形式同时作为账户目录、管理页面标识和日 生成服务启动时与之后每 10 分钟删除各账户到期的资源：上传文件在 `expires_at` 之后，内联附件闲置 1 小时且没有进行中的请求引用之后，生成视频与其任务自完成起 48 小时（没有完成的任务自创建起 48 小时），Live 续接凭证自最近一次发放起 24 小时，旧版本留下的请求临时文件（以 `drive-file` 记录、名为 `inline-media-<序号>` 的内联附件，Omni 输入图片，以及没有摘要、其他账户有更早上传的同名同大小文件的跨账户副本）自上传起 48 小时。Drive 中的文件先删除再解除绑定，Drive 返回 404 时按已删除处理，其他删除失败保留绑定、下一轮重试；没有空闲并发槽位的账户在下一轮回收；其余没有 `expires_at` 的上传文件不按时间删除。 -每次运行状态事务先取得 `auth/.leases/<账户>.runtime.lock`，每 25ms 尝试一次，等待上限为 2 秒；携带调用方 context（Go 取消上下文）的事务在调用方取消或 deadline 更早到达时立即结束。取得锁后重新读取当前磁盘值，只合并目标字段，以临时文件原子替换 `runtime-state.json`，再同步内存状态与资源归属。锁等待失败、读取失败、写回失败和 unlock 失败均返回原始错误链。 +每次运行状态事务先取得 `auth/.leases/<账户>.runtime.lock`，每 25ms 尝试一次，直到锁释放；携带调用方 context（Go 取消上下文）的事务在调用方取消或 deadline 到达时结束等待。取得锁后重新读取当前磁盘值，只合并目标字段，以临时文件原子替换 `runtime-state.json`，再同步内存状态与资源归属。锁等待失败、读取失败、写回失败和 unlock 失败均返回原始错误链。 调度器按实时模型方法、capability、AccessModes 和账户权益选择候选，优先使用已经成功调用目标 scope（模型资格与冷却的状态记录范围）的预热账户，再按目标模型最近一次真实首事件耗时排序。常驻 Worker 优先覆盖可调用模型更多的账户。预热数量低于上限时提升待机账户，预热账户均忙时等待并发槽位。同账号 WAA proof 串行生成，已准备的 MakerSuite HTTP 请求并发执行；首个活动请求获取 `.leases/<邮箱>.lock`，最后一个释放。Drive file、视频任务与结果 file 始终使用创建账户。 @@ -246,7 +246,7 @@ Waa/Create -> send protected MakerSuite RPC ``` -`WAA_BACKEND=camoufox` 时官方 VM 运行在账户固定指纹的 Camoufox 页面，受保护请求由页面 `XMLHttpRequest` 发送；最后一轮是带文本的 user 轮次的 `GenerateContent` 由页面应用在输入框写入该轮文本并点击 Run 时发出（与页面上一轮的对话相同时点击 Rerun），proof 由页面应用以服务给出的 binding 摘要生成，发送时正文换成服务编码的请求正文；`WAA_BACKEND=spidermonkey` 时 VM 运行在服务进程内的 SpiderMonkey 与 Firefox 形状宿主中，受保护请求由 Go HTTP 以 Firefox 152 网络形状发送。 +`WAA_BACKEND=camoufox` 时官方 VM 运行在账户固定指纹的 Camoufox 页面，受保护请求由页面 `XMLHttpRequest` 发送；最后一轮是带文本的 user 轮次的 `GenerateContent` 首轮由页面应用在输入框写入该轮文本并点击 Run 时发出（与页面上一轮的对话相同时点击 Rerun），proof 由页面应用以服务给出的 binding 摘要生成，发送时正文换成服务编码的请求正文，自动续写的后续轮次沿用页面写入首轮的 proof 直接发送；`WAA_BACKEND=spidermonkey` 时 VM 运行在服务进程内的 SpiderMonkey 与 Firefox 形状宿主中，受保护请求由 Go HTTP 以 Firefox 152 网络形状发送。 `Waa/Create` 请求是 JSON+protobuf 数组。Worker 启动时只带 request key，VM 刷新时追加当前 interpreter hash 与上一 VM 的无绑定 snapshot： @@ -751,10 +751,10 @@ ListModels 能力码 91（目录能力 `output_continuation`）的模型由官 1. 请求的 generation config field 46 为 `1`，根 field 11 省略；max output 大于 32768 时首轮发送 32768 2. 候选 field 9（JSON 索引 8）给出续写 token。完成帧以 `max_tokens` 结束且同一帧带 token、本轮上限小于请求的输出总量，或以终止原因 22 结束时，本轮不形成终态 -3. 下一轮复制原请求，根消息 field 17 写入最近的 token，max output 为 `min(输出总量, (轮次 + 1) × 32768)`，轮次从 0 计 +3. 下一轮复制原请求，沿用首轮写入的 field 5 proof，根消息 field 17 写入最近的 token，max output 为 `min(输出总量, (轮次 + 1) × 32768)`，轮次从 0 计 4. 终止原因 22 而响应没有 token 时请求失败 -各轮正文、思考与工具事件按顺序输出为一个响应，终态取最后一轮。usage 以最后一轮为基础，visible output、thought 与 tool tokens 逐轮累加，total 为最后一轮 total 加之前各轮的这三项。未指定输出上限时输出总量为模型的 `inputTokenLimit`。输出总量超过 32768 的自动续写请求只经 Playground 调度，未指定输出上限且含 File Search、`serviceTier`、`store:false` 等经 Build 通道发送的设置时仍可经 Build，按 Build 的单轮上限发送；请求期限为 `REQUEST_TIMEOUT`，每个续写轮开始时重新计时。每个续写轮在日志中记一条“上游调用”，模式后附 `续写=<轮次>`。 +各轮正文、思考与工具事件按顺序输出为一个响应，终态取最后一轮。usage 以最后一轮为基础，visible output、thought 与 tool tokens 逐轮累加，total 为最后一轮 total 加之前各轮的这三项。未指定输出上限时输出总量为模型的 `inputTokenLimit`。输出总量超过 32768 的自动续写请求只经 Playground 调度，未指定输出上限且含 File Search、`serviceTier`、`store:false` 等经 Build 通道发送的设置时仍可经 Build，按 Build 的单轮上限发送；请求期限为 `REQUEST_TIMEOUT`，每个续写轮开始时重新计时。每个续写轮在日志中记一条“上游调用”，模式后附 `续写=<轮次>`。续写 token 可由其他账户接续：续写轮以 401、403、429 或 5xx 失败时，未指定账户、不含文件与内联媒体的请求换到本请求续写轮尚未失败的其他账户，以该账户的新 proof 重发同一续写轮，之后各轮沿用这个 proof；额度失败的账户按额度冷却规则冷却，用量明细记为最后生成的账户。 ### Finish 与错误 diff --git a/internal/aistudio/accounts.go b/internal/aistudio/accounts.go index 0b647e8b..9b6aea88 100644 --- a/internal/aistudio/accounts.go +++ b/internal/aistudio/accounts.go @@ -40,7 +40,6 @@ const ( globalCooldownKey = "*" externalLeasePoll = 100 * time.Millisecond runtimeLockPoll = 25 * time.Millisecond - runtimeLockLimit = 2 * time.Second ) // AccountState 表示账户当前是否可调度 @@ -3367,9 +3366,7 @@ func lockRuntimeState(ctx context.Context, account *Account) (*flock.Flock, erro return nil, fmt.Errorf("创建账户状态锁目录: %w", err) } lock := flock.New(filepath.Join(leaseDirectory, filepath.Base(account.Directory)+".runtime.lock")) - lockCtx, cancel := context.WithTimeout(ctx, runtimeLockLimit) - defer cancel() - _, err := lock.TryLockContext(lockCtx, runtimeLockPoll) + _, err := lock.TryLockContext(ctx, runtimeLockPoll) if err != nil { return nil, fmt.Errorf("锁定账户运行状态: %w", err) } diff --git a/internal/aistudio/build_embed.go b/internal/aistudio/build_embed.go index ceaeeb2f..a44a43ad 100644 --- a/internal/aistudio/build_embed.go +++ b/internal/aistudio/build_embed.go @@ -136,7 +136,7 @@ func (t *WorkerProtectedTransport) DoProtectedBuild(ctx context.Context, selecti if err != nil { return nil, err } - return t.doPrepared(ctx, prompt, page, buildProofField, selection, rpc) + return t.doPrepared(ctx, prompt, page, buildProofField, "", selection, rpc) } // buildEmbedBatchLimit 是上游 batchEmbedContents 单次接受的最大请求数 diff --git a/internal/aistudio/client.go b/internal/aistudio/client.go index 75d04ee0..029610e7 100644 --- a/internal/aistudio/client.go +++ b/internal/aistudio/client.go @@ -35,11 +35,12 @@ type RPCRequest struct { Streaming bool } -// RPCResponse 描述认证传输层返回的响应头和实时正文 +// RPCResponse 描述认证传输层返回的响应头和实时正文，Proof 为受保护请求写入的 WAA proof type RPCResponse struct { StatusCode int Header http.Header Body io.ReadCloser + Proof string } // RPCTransport 负责认证、账户租约、Cookie 写回和真实网络发送 diff --git a/internal/aistudio/generate.go b/internal/aistudio/generate.go index 294d851f..4954ddf1 100644 --- a/internal/aistudio/generate.go +++ b/internal/aistudio/generate.go @@ -1168,12 +1168,24 @@ func (c *Client) sendPlayground(ctx context.Context, request GenerateRequest, en if err != nil { return nil, nil, err } + request.continuationProof = response.Proof return response, func(source io.Reader, emit func(Event) error) error { return c.decodePlaygroundRounds(ctx, request, entry.defaults, runtime, source, restoreFunctionNames(request.Tools.Functions, emit)) }, nil } -// decodePlaygroundRounds 解码 GenerateContent 流，上游要求续写时携带续写 token 发送下一轮，各轮输出依次转发 +// continuationAccountsKey 是上下文中续写轮换账户函数的键 +type continuationAccountsKey struct{} + +// ContinuationAccounts 在续写轮以可换账户的错误失败后返回改由其他账户发送该轮的上下文 +type ContinuationAccounts func(context.Context, error) (context.Context, error) + +// ContextWithContinuationAccounts 返回续写轮可经 accounts 换账户发送的上下文 +func ContextWithContinuationAccounts(ctx context.Context, accounts ContinuationAccounts) context.Context { + return context.WithValue(ctx, continuationAccountsKey{}, accounts) +} + +// decodePlaygroundRounds 解码 GenerateContent 流，上游要求续写时携带续写 token 发送下一轮，续写轮以可换账户的错误失败时经上下文中的 ContinuationAccounts 换账户重发，各轮输出依次转发 func (c *Client) decodePlaygroundRounds(ctx context.Context, request GenerateRequest, defaults GenerationDefaults, runtime RequestContext, source io.Reader, emit func(Event) error) error { total := int64(0) if defaults.ContinuedOutputTokens > 0 { @@ -1233,9 +1245,34 @@ func (c *Client) decodePlaygroundRounds(ctx context.Context, request GenerateReq } closeRound() response, err := c.doProtected(ctx, request, body) + for err != nil && retryableAccountError(err) { + accounts, ok := ctx.Value(continuationAccountsKey{}).(ContinuationAccounts) + if !ok { + break + } + next, switchErr := accounts(ctx, err) + if switchErr != nil { + err = errors.Join(err, switchErr) + break + } + ctx = next + lease, _ := AccountLeaseFromContext(ctx) + request.AccountID, request.continuationProof = lease.Account().ID, "" + if c.contextProvider != nil { + if runtime, err = c.contextProvider.RequestContext(ctx, request.AccountID); err != nil { + return fmt.Errorf("读取 AI Studio 请求上下文: %w", err) + } + } + if body, err = EncodeGenerateContentRequest(request, defaults, runtime); err != nil { + return err + } + reportUpstreamMode(ctx, "GenerateContent", fmt.Sprintf("stream | 续写=%d", request.continuationRound), "") + response, err = c.doProtected(ctx, request, body) + } if err != nil { return err } + request.continuationProof = response.Proof stopClose := context.AfterFunc(ctx, func() { _ = response.Body.Close() }) diff --git a/internal/aistudio/imagen.go b/internal/aistudio/imagen.go index ec435eea..f33987f7 100644 --- a/internal/aistudio/imagen.go +++ b/internal/aistudio/imagen.go @@ -81,7 +81,7 @@ func (c *Client) GenerateImage(ctx context.Context, request ImageRequest) ([]Gen // DoProtectedImage 写入 GenerateImage fresh proof 后经 Worker 页面发送请求，binding 与页面输入框均为提示词 func (t *WorkerProtectedTransport) DoProtectedImage(ctx context.Context, request ImageRequest, rpc RPCRequest) (*RPCResponse, error) { - return t.doPrepared(ctx, request.Prompt, request.Prompt, 3, AccountSelection{ + return t.doPrepared(ctx, request.Prompt, request.Prompt, 3, "", AccountSelection{ ModelID: request.Model, Method: "predict", AccountID: request.AccountID, }, rpc) } diff --git a/internal/aistudio/runtime_native.go b/internal/aistudio/runtime_native.go index 9240bc4e..d02d151c 100644 --- a/internal/aistudio/runtime_native.go +++ b/internal/aistudio/runtime_native.go @@ -96,17 +96,17 @@ type pageGenerator interface { Generate(context.Context, camoufoxnative.GenerateTurn) (*camoufoxnative.ProtectedResponse, error) } -// generator 返回在页面界面运行该请求的 runtime：Camoufox 后端、GenerateContent 且最后一个轮次是带文本的 user 轮次 +// generator 返回在页面界面运行该请求的 runtime：Camoufox 后端、未带 proof 的 GenerateContent 且最后一个轮次是带文本的 user 轮次 func (worker *NativeWorker) generator(request ProtectedRequest) (pageGenerator, bool) { generator, ok := worker.current().(pageGenerator) - if !ok || !strings.HasSuffix(request.URL, "/GenerateContent") { + if !ok || request.Proof != "" || !strings.HasSuffix(request.URL, "/GenerateContent") { return nil, false } _, typed := camoufoxnative.GeneratePrompt(request.Body) return generator, typed } -// Prepare 生成 fresh proof 并写入请求指定的 WAA field；在页面界面运行的 GenerateContent 由页面应用取 proof，只返回公共协议头 +// Prepare 生成 fresh proof 并写入请求指定的 WAA field，请求带 Proof 时写入该 proof；在页面界面运行的 GenerateContent 由页面应用取 proof，只返回公共协议头 func (worker *NativeWorker) Prepare(ctx context.Context, request ProtectedRequest) (PreparedProtectedRequest, error) { start, end, err := arrayElementSpan(request.Body, request.ProofField-1) if err != nil { @@ -119,14 +119,20 @@ func (worker *NativeWorker) Prepare(ctx context.Context, request ProtectedReques } return PreparedProtectedRequest{Body: request.Body, Headers: headers}, nil } - proof, headers, err := worker.proofAndHeaders(ctx, request.Prompt, request.PagePrompt, strings.HasSuffix(request.URL, "/GenerateContent") && request.PagePrompt != "") + proof := request.Proof + var headers http.Header + if proof == "" { + proof, headers, err = worker.proofAndHeaders(ctx, request.Prompt, request.PagePrompt, strings.HasSuffix(request.URL, "/GenerateContent") && request.PagePrompt != "") + } else { + headers, err = worker.current().ProtocolHeaders(ctx) + } if err != nil { return PreparedProtectedRequest{}, err } encoded, _ := json.Marshal(proof) body := make([]byte, 0, len(request.Body)-(end-start)+len(encoded)) body = append(append(append(body, request.Body[:start]...), encoded...), request.Body[end:]...) - return PreparedProtectedRequest{Body: body, Headers: headers}, nil + return PreparedProtectedRequest{Body: body, Headers: headers, Proof: proof}, nil } // pageRunner 是写入提示词并点击 Run 后取 GenerateContent proof 的 workerRuntime @@ -216,7 +222,7 @@ func arrayElementSpan(body []byte, index int) (int, int, error) { // pageCountTokensDelay 是官网页面在 GenerateContent 发出后统计用户轮次 token 的时延 const pageCountTokensDelay = 2200 * time.Millisecond -// SendProtected 经账户 WAA runtime 流式发送已准备的请求：Camoufox 后端的 GenerateContent 在页面界面运行，请求头、proof 与随后的 CountTokens 都由页面应用产生；其余 GenerateContent 带官网点击运行时的 Priority: u=0，发出 pageCountTokensDelay 后由 runtime 以相同请求头统计页面提示词的 token +// SendProtected 经账户 WAA runtime 流式发送已准备的请求，响应带实际写入的 proof：Camoufox 后端未带 proof 的 GenerateContent 在页面界面运行，请求头、proof 与随后的 CountTokens 都由页面应用产生；其余 GenerateContent 带官网点击运行时的 Priority: u=0，发出 pageCountTokensDelay 后由 runtime 以相同请求头统计页面提示词的 token func (worker *NativeWorker) SendProtected(ctx context.Context, request ProtectedRequest) (*RPCResponse, error) { if generator, ok := worker.generator(request); ok { return worker.generate(ctx, generator, request) @@ -244,6 +250,7 @@ func (worker *NativeWorker) SendProtected(ctx context.Context, request Protected StatusCode: response.StatusCode, Header: response.Header, Body: response.Body, + Proof: request.Proof, }, nil } @@ -265,7 +272,7 @@ func (worker *NativeWorker) generate(ctx context.Context, generator pageGenerato return nil, err } worker.setPhase(WorkerReady) - return &RPCResponse{StatusCode: response.StatusCode, Header: response.Header, Body: response.Body}, nil + return &RPCResponse{StatusCode: response.StatusCode, Header: response.Header, Body: response.Body, Proof: response.Proof}, nil } // countTokensBody 按官网页面对用户轮次的统计，以 GenerateContent 的模型构造页面提示词的 CountTokens 正文；不是带页面提示词的 GenerateContent 时返回 false diff --git a/internal/aistudio/service.go b/internal/aistudio/service.go index 084004ed..df789bef 100644 --- a/internal/aistudio/service.go +++ b/internal/aistudio/service.go @@ -123,21 +123,22 @@ func (t *WorkerProtectedTransport) DoProtected(ctx context.Context, request Gene page = "" } modelID := strings.TrimPrefix(strings.TrimSpace(request.Model), "models/") - return t.doPrepared(ctx, prompt, page, proofField, AccountSelection{ + return t.doPrepared(ctx, prompt, page, proofField, request.continuationProof, AccountSelection{ ModelID: modelID, Method: "generateContent", AccountID: strings.TrimSpace(request.AccountID), }, rpc) } -// doPrepared 为请求写入 proof 后经账户的 WAA 运行时发送，并写回运行时导出的 Cookie +// doPrepared 为请求写入 proof 后经账户的 WAA 运行时发送，并写回运行时导出的 Cookie；proof 非空时写入该 proof func (t *WorkerProtectedTransport) doPrepared( ctx context.Context, prompt string, page string, proofField int, + proof string, selection AccountSelection, rpc RPCRequest, ) (*RPCResponse, error) { - lease, worker, rpc, err := t.prepareProtectedRequest(ctx, prompt, page, proofField, selection, rpc) + lease, worker, rpc, proof, err := t.prepareProtectedRequest(ctx, prompt, page, proofField, proof, selection, rpc) if err != nil { return nil, err } @@ -156,7 +157,7 @@ func (t *WorkerProtectedTransport) doPrepared( } reportRequestPhase(ctx, RequestPhaseSendingUpstream) response, err := worker.SendProtected(ctx, ProtectedRequest{ - URL: rpc.URL, Headers: headers, Body: rpc.Body, Prompt: prompt, PagePrompt: page, ProofField: proofField, + URL: rpc.URL, Headers: headers, Body: rpc.Body, Prompt: prompt, PagePrompt: page, ProofField: proofField, Proof: proof, }) if err != nil { return nil, err @@ -197,30 +198,31 @@ func (t *WorkerProtectedTransport) prepareProtectedRequest( prompt string, page string, proofField int, + proof string, selection AccountSelection, rpc RPCRequest, -) (*AccountLease, ProtectedPreparer, RPCRequest, error) { +) (*AccountLease, ProtectedPreparer, RPCRequest, string, error) { lease, ok := AccountLeaseFromContext(ctx) if !ok { - return nil, nil, RPCRequest{}, fmt.Errorf("受保护请求缺少账户租约") + return nil, nil, RPCRequest{}, "", fmt.Errorf("受保护请求缺少账户租约") } if err := validateLeaseSelection(lease, selection); err != nil { - return nil, nil, RPCRequest{}, err + return nil, nil, RPCRequest{}, "", err } worker, err := t.workers.Worker(ctx, lease.Account().ID, selection.ModelID) if err != nil { - return nil, nil, RPCRequest{}, fmt.Errorf("获取账户 WAA preparer: %w", err) + return nil, nil, RPCRequest{}, "", fmt.Errorf("获取账户 WAA preparer: %w", err) } reportRequestPhase(ctx, RequestPhasePreparingWAA) prepared, err := worker.Prepare(ctx, ProtectedRequest{ URL: rpc.URL, Headers: rpc.Header.Clone(), Body: rpc.Body, - Prompt: prompt, PagePrompt: page, ProofField: proofField, + Prompt: prompt, PagePrompt: page, ProofField: proofField, Proof: proof, }) if err != nil { - return nil, nil, RPCRequest{}, fmt.Errorf("准备 fresh WAA proof: %w", err) + return nil, nil, RPCRequest{}, "", fmt.Errorf("准备 fresh WAA proof: %w", err) } if prepared.Headers == nil || len(prepared.Body) == 0 { - return nil, nil, RPCRequest{}, fmt.Errorf("WAA preparer 返回空请求") + return nil, nil, RPCRequest{}, "", fmt.Errorf("WAA preparer 返回空请求") } requestHeaders := rpc.Header rpc.AccountID = lease.Account().ID @@ -233,12 +235,12 @@ func (t *WorkerProtectedTransport) prepareProtectedRequest( } } rpc.Header.Set("Content-Type", JSONProtobufContentType) - return lease, worker, rpc, nil + return lease, worker, rpc, prepared.Proof, nil } // DoProtectedTitle 写入 GenerateTitle fresh proof 后经 Worker 页面发送请求，binding 为提示词与回答以空格连接，页面输入框为空 func (t *WorkerProtectedTransport) DoProtectedTitle(ctx context.Context, request TitleRequest, rpc RPCRequest) (*RPCResponse, error) { - return t.doPrepared(ctx, request.Prompt+" "+request.Answer, "", 3, AccountSelection{ModelID: request.Model}, rpc) + return t.doPrepared(ctx, request.Prompt+" "+request.Answer, "", 3, "", AccountSelection{ModelID: request.Model}, rpc) } func bindingPrompt(request GenerateRequest) (string, error) { diff --git a/internal/aistudio/types.go b/internal/aistudio/types.go index 68bc070c..641cff76 100644 --- a/internal/aistudio/types.go +++ b/internal/aistudio/types.go @@ -452,6 +452,8 @@ type GenerateRequest struct { continuationRound int // continuationToken 为上一轮响应给出的续写 token continuationToken string + // continuationProof 为首轮请求写入的 WAA proof，各续写轮沿用 + continuationProof string // restoreModelTurns 表示 Playground 请求按本进程保存的生成结果补回历史模型轮 restoreModelTurns bool // turnToken 为历史中最后一个模型轮的轮次令牌 diff --git a/internal/aistudio/waa.go b/internal/aistudio/waa.go index 7c739f82..f73272d6 100644 --- a/internal/aistudio/waa.go +++ b/internal/aistudio/waa.go @@ -6,7 +6,7 @@ import ( "time" ) -// ProtectedRequest 表示需要 WAA 保护的 AI Studio 请求 +// ProtectedRequest 表示需要 WAA 保护的 AI Studio 请求，Proof 非空时写入该 proof 并直接发送 type ProtectedRequest struct { URL string Headers http.Header @@ -14,6 +14,7 @@ type ProtectedRequest struct { Prompt string PagePrompt string ProofField int + Proof string } // followUp 是受保护请求发出 Delay 后以相同请求头发出的伴随请求，响应被丢弃 @@ -37,10 +38,11 @@ func followUpFrom(ctx context.Context) (followUp, bool) { return item, ok } -// PreparedProtectedRequest 表示已写入 fresh proof 的请求 +// PreparedProtectedRequest 表示已写入 proof 的请求与写入的 proof，由页面界面运行的请求 proof 为空 type PreparedProtectedRequest struct { Body []byte Headers http.Header + Proof string } // WorkerPhase 表示账户 runtime 当前阶段 diff --git a/internal/app/runtime.go b/internal/app/runtime.go index d4217cd8..ccd700c7 100644 --- a/internal/app/runtime.go +++ b/internal/app/runtime.go @@ -1728,6 +1728,7 @@ type generateAttempt struct { lease *aistudio.AccountLease selection aistudio.AccountSelection hedge bool + unbound bool accountLabel string proxy string workerGeneration uint64 @@ -1740,6 +1741,8 @@ type generateAttempt struct { err error cancel context.CancelFunc done chan struct{} + // serving 是正在生成回答的账户租约，续写轮换账户后为新账户的租约 + serving atomic.Pointer[aistudio.AccountLease] } // generateRetry 是生成请求跨尝试的换号与恢复状态 @@ -2009,7 +2012,7 @@ func (service *trackedService) generateWithRetry( } service.forwardEvents( clientCtx, requestCtx, func() { winner.cancel(); cancel() }, request.ID, - winner.first, winner.source, destination, winner.lease, winner.copies, winner.activity, modelID, + winner.first, winner.source, destination, &winner.serving, winner.copies, winner.activity, modelID, ) } @@ -2043,12 +2046,14 @@ func (service *trackedService) startAttempt( selection aistudio.AccountSelection, hedge bool, ) *generateAttempt { + unbound := strings.TrimSpace(request.AccountID) == "" && resourceID == "" request.AccountID = lease.Account().ID attempt := &generateAttempt{ - lease: lease, selection: selection, hedge: hedge, accountLabel: lease.Account().Config.Label, + lease: lease, selection: selection, hedge: hedge, unbound: unbound, accountLabel: lease.Account().Config.Label, proxy: service.proxies.Binding(request.AccountID), workerGeneration: service.workers.WorkerGeneration(request.AccountID), startedAt: time.Now(), activity: &upstreamActivity{}, done: make(chan struct{}), } + attempt.serving.Store(lease) attemptCtx, cancel := context.WithCancel(requestCtx) attempt.cancel = cancel go func() { @@ -2081,18 +2086,7 @@ func (service *trackedService) runAttempt( } originalContents := request.Contents ctx = aistudio.ContextWithAccountLease(ctx, lease) - ctx = aistudio.ContextWithUpstreamModeObserver(ctx, func(method, mode, reason string) { - level := "INFO" - message := fmt.Sprintf("上游调用 | 通道=%s | 模式=%s | RPC=%s", lease.Channel(), mode, method) - if reason != "" && service.pool.BuildEnabled() { - level = "WARN" - if fallbackReason != "" { - reason = fallbackReason + "; " + reason - } - message += " | 回退流式 | 原因=" + reason - } - service.requests.logRequestProgress(request.ID, accountLabel, level, message) - }) + ctx = aistudio.ContextWithUpstreamModeObserver(ctx, service.upstreamModeObserver(request.ID, lease, fallbackReason)) var err error copiedFileCount := 0 if resourceID != "" { @@ -2104,6 +2098,12 @@ func (service *trackedService) runAttempt( } } mediaCount, mediaBytes := inlineMediaInput(request.Contents) + if attempt.unbound && mediaCount == 0 { + failed := make(map[string]struct{}) + ctx = aistudio.ContextWithContinuationAccounts(ctx, func(roundCtx context.Context, err error) (context.Context, error) { + return service.continuationAccount(roundCtx, request.ID, modelID, attempt, failed, err) + }) + } if err == nil && mediaCount > 0 { uploadStartedAt := time.Now() request.Contents, attempt.copies, err = service.service.(interface { @@ -2169,6 +2169,73 @@ func (service *trackedService) runAttempt( attempt.err = err } +// upstreamModeObserver 返回把租约账户的上游调用方式记入请求日志的观察函数，回退流式时附上 fallbackReason +func (service *trackedService) upstreamModeObserver(requestID string, lease *aistudio.AccountLease, fallbackReason string) func(string, string, string) { + accountLabel := lease.Account().Config.Label + return func(method, mode, reason string) { + level := "INFO" + message := fmt.Sprintf("上游调用 | 通道=%s | 模式=%s | RPC=%s", lease.Channel(), mode, method) + if reason != "" && service.pool.BuildEnabled() { + level = "WARN" + if fallbackReason != "" { + reason = fallbackReason + "; " + reason + } + message += " | 回退流式 | 原因=" + reason + } + service.requests.logRequestProgress(requestID, accountLabel, level, message) + } +} + +// continuationAccount 把续写轮在当前账户上的失败记为一次尝试，额度失败时冷却该账户，释放其租约后从未在续写轮失败过的其余账户取得租约，返回改由该账户发送续写轮的上下文 +func (service *trackedService) continuationAccount(ctx context.Context, requestID string, modelID string, attempt *generateAttempt, failed map[string]struct{}, err error) (context.Context, error) { + current := attempt.serving.Load() + label := current.Account().Config.Label + api.AddAccessLogAttempt(ctx, api.RequestAttempt{Account: label, Channel: string(current.Channel()), Error: err.Error()}) + failed[current.Account().ID] = struct{}{} + if cooldown, quotaFailure := aistudio.QuotaCooldownForError(err, time.Now()); quotaFailure { + if _, stateErr := service.markQuotaCooldown(requestID, modelID, current, cooldown); stateErr != nil { + return nil, stateErr + } + } + if releaseErr := current.Release(); releaseErr != nil { + return nil, releaseErr + } + selection := aistudio.AccountSelection{ModelID: modelID, Method: "generateContent", PlaygroundOnly: true, AllowedAccountIDs: []string{}} + enabled, _ := service.pool.EnabledAccounts() + for _, accountID := range enabled { + if _, exists := failed[accountID]; !exists { + selection.AllowedAccountIDs = append(selection.AllowedAccountIDs, accountID) + } + } + lease, acquireErr := service.acquireWarmLease(ctx, selection, false) + if acquireErr != nil { + return nil, acquireErr + } + attempt.serving.Store(lease) + service.requests.logRequestProgress(requestID, label, "WARN", fmt.Sprintf( + "账号切换 | 续写轮 | 模型=%s\n原因: %s", modelID, strings.TrimSpace(err.Error()), + )) + service.markAttemptAccount(ctx, requestID, modelID, lease) + next := aistudio.ContextWithAccountLease(ctx, lease) + return aistudio.ContextWithUpstreamModeObserver(next, service.upstreamModeObserver(requestID, lease, "")), nil +} + +// markQuotaCooldown 按额度失败冷却租约账户在该模型或全局范围上的调度并记入请求日志，返回实际生效的恢复时刻 +func (service *trackedService) markQuotaCooldown(requestID, modelID string, lease *aistudio.AccountLease, cooldown aistudio.QuotaCooldown) (time.Time, error) { + scope := lease.CooldownScope(modelID) + scopeLabel := scope + if cooldown.Global { + scope, scopeLabel = "", "全局" + } + until, err := service.pool.MarkQuotaCooldownIfGeneration(lease.Account().ID, scope, lease.ModelAccessGeneration(), lease.CheckedAt(), cooldown) + if err == nil && !until.IsZero() { + service.requests.logRequestProgress(requestID, lease.Account().Config.Label, "WARN", fmt.Sprintf( + "账号冷却 | 类型=%s | 范围=%s | 恢复=%s", cooldown.Kind, scopeLabel, until.Local().Format(time.RFC3339), + )) + } + return until, err +} + // quotaCooldownError 是带恢复时刻的额度失败，恢复时刻取本次请求中额度失败账户最早的冷却恢复时刻 type quotaCooldownError struct { err error @@ -2230,15 +2297,7 @@ func (service *trackedService) settleAttempt( } cooldown, quotaFailure := aistudio.QuotaCooldownForError(err, time.Now()) if quotaFailure { - modelAccessScope := lease.CooldownScope(modelID) - scopeLabel := modelAccessScope - if cooldown.Global { - modelAccessScope = "" - scopeLabel = "全局" - } - until, stateErr := service.pool.MarkQuotaCooldownIfGeneration( - accountID, modelAccessScope, lease.ModelAccessGeneration(), lease.CheckedAt(), cooldown, - ) + until, stateErr := service.markQuotaCooldown(requestID, modelID, lease, cooldown) if stateErr != nil { err = errors.Join(err, stateErr) retryable = false @@ -2248,10 +2307,6 @@ func (service *trackedService) settleAttempt( retry.quotaUntil = until } err = "aCooldownError{err: err, until: retry.quotaUntil} - service.requests.logRequestProgress(requestID, accountLabel, "WARN", fmt.Sprintf( - "账号冷却 | 类型=%s | 范围=%s | 恢复=%s", - cooldown.Kind, scopeLabel, until.Local().Format(time.RFC3339), - )) } if !cooldown.Global && service.pool.AccountChannelAvailable(accountID, attempt.selection) { delete(retry.attempted, accountID) @@ -2286,7 +2341,7 @@ func (service *trackedService) settleAttempt( if !quotaFailure && !buildRejected { retry.attemptFailure = err } - if releaseErr := lease.Release(); releaseErr != nil { + if releaseErr := attempt.serving.Load().Release(); releaseErr != nil { return attemptOutcome{err: errors.Join(err, releaseErr), accountID: accountID, accountLabel: accountLabel} } recoverWorker := false @@ -2526,7 +2581,7 @@ func (service *trackedService) forwardEvents( first aistudio.Event, source <-chan aistudio.Event, destination chan<- aistudio.Event, - lease *aistudio.AccountLease, + serving *atomic.Pointer[aistudio.AccountLease], temporaryCopies *aistudio.TemporaryFileCopies, activity *upstreamActivity, requestedModelID string, @@ -2534,8 +2589,7 @@ func (service *trackedService) forwardEvents( state := "completed" var requestErr error terminal := false - accountLabel := lease.Account().Config.Label - accessGeneration := lease.ModelAccessGeneration() + accountLabel := func() string { return serving.Load().Account().Config.Label } modelID := strings.TrimPrefix(strings.TrimSpace(first.ProviderModel), "models/") var lastEventAt time.Time lastEventKind := "-" @@ -2565,10 +2619,10 @@ func (service *trackedService) forwardEvents( api.SetAccessLogGenerationResult(requestCtx, usage, toolCalls) if temporaryCopies != nil { if err := temporaryCopies.Cleanup(); err != nil { - service.requests.log(accountLabel, "WARN", "临时文件清理失败 | 错误="+err.Error()) + service.requests.log(accountLabel(), "WARN", "临时文件清理失败 | 错误="+err.Error()) } } - if err := lease.Release(); err != nil { + if err := serving.Load().Release(); err != nil { state = "failed" requestErr = errors.Join(requestErr, err) if clientCtx.Err() == nil { @@ -2597,7 +2651,7 @@ func (service *trackedService) forwardEvents( case <-stall: stalled = true stallTimer.Reset(waitReportInterval) - service.requests.logRequestProgress(requestID, accountLabel, "WARN", fmt.Sprintf( + service.requests.logRequestProgress(requestID, accountLabel(), "WARN", fmt.Sprintf( "事件流停顿 | 模型=%s | 已等待=%s | 最近事件=%s | 推理=%d | 正文=%d | %s", modelID, time.Since(lastEventAt).Round(time.Second), lastEventKind, reasoningEvents, contentEvents, activity.logFields(time.Now()), @@ -2623,7 +2677,7 @@ func (service *trackedService) forwardEvents( } now := time.Now() if stalled { - service.requests.logRequestProgress(requestID, accountLabel, "INFO", fmt.Sprintf( + service.requests.logRequestProgress(requestID, accountLabel(), "INFO", fmt.Sprintf( "事件流恢复 | 模型=%s | 停顿=%s | 当前事件=%s", modelID, now.Sub(lastEventAt).Round(time.Millisecond), event.Kind, )) @@ -2641,14 +2695,14 @@ func (service *trackedService) forwardEvents( case aistudio.EventToolCall: toolCalls++ } - api.SetAccessLogTarget(requestCtx, event.ProviderModel, lease.Account().Config.Label) + api.SetAccessLogTarget(requestCtx, event.ProviderModel, accountLabel()) if terminal { continue } if event.Kind == aistudio.EventError { requestErr = event.Err if aistudio.DefinitiveAuthenticationFailure(event.Err) { - if stateErr := lease.ConfirmAuthenticationFailure(requestCtx, event.Err); stateErr != nil { + if stateErr := serving.Load().ConfirmAuthenticationFailure(requestCtx, event.Err); stateErr != nil { requestErr = errors.Join(requestErr, stateErr) event.Err = requestErr } @@ -2675,8 +2729,9 @@ func (service *trackedService) forwardEvents( } if !verified && event.Kind == aistudio.EventFinish { verified = true + lease := serving.Load() service.markModelAccessVerifiedAsync( - lease.Account().ID, accountLabel, lease.CooldownScope(requestedModelID), accessGeneration, lease.CheckedAt(), + lease.Account().ID, lease.Account().Config.Label, lease.CooldownScope(requestedModelID), lease.ModelAccessGeneration(), lease.CheckedAt(), ) } } diff --git a/internal/camoufoxnative/generate.go b/internal/camoufoxnative/generate.go index 61ca726c..73ebdb9a 100644 --- a/internal/camoufoxnative/generate.go +++ b/internal/camoufoxnative/generate.go @@ -122,7 +122,7 @@ func nextChatAction(previous, contents []json.RawMessage) chatAction { return chatNew } -// generateArmFunction 是在页面上为下一次 Run 安装一次性钩子的函数声明：官网高层 proof 函数下一次被调用时记下请求已开始并以本请求的 digest 取 proof，proof 返回后下一个 GenerateContent XHR 发送时把正文换成 head + proof + tail，并在该 XHR 上监听响应，经通道先发出 {sent: true}，再依次发出响应头、正文块与结束标记；参数依次为页面对象、请求 ID、digest、head、tail 与通道 +// generateArmFunction 是在页面上为下一次 Run 安装一次性钩子的函数声明：官网高层 proof 函数下一次被调用时记下请求已开始并以本请求的 digest 取 proof，proof 返回后下一个 GenerateContent XHR 发送时把正文换成 head + proof + tail，并在该 XHR 上监听响应，经通道先发出带 proof 的 {sent: true, proof}，再依次发出响应头、正文块与结束标记；参数依次为页面对象、请求 ID、digest、head、tail 与通道 const generateArmFunction = `(page, id, digest, head, tail, emit) => { const makerSuite = window.default_MakerSuite; const open = XMLHttpRequest.prototype.open; @@ -161,7 +161,7 @@ const generateArmFunction = `(page, id, digest, head, tail, emit) => { changed(); request.resume?.(); }); - emit(JSON.stringify({sent: true})); + emit(JSON.stringify({sent: true, proof})); return send.call(xhr, head + JSON.stringify(proof) + tail); }; } @@ -217,7 +217,7 @@ func (worker *Worker) generate(ctx context.Context, tab *chatTab, turn GenerateT unboost() return nil, worker.abandonProtectedRequest(ctx, tab.contextID, tab.page, requestID, err) } - err = worker.waitSent(ctx, tab, requestID, channel) + proof, err := worker.waitSent(ctx, tab, requestID, channel) unboost() if err != nil { return nil, worker.abandonProtectedRequest(ctx, tab.contextID, tab.page, requestID, fmt.Errorf("等待页面应用发出 GenerateContent: %w", err)) @@ -239,6 +239,7 @@ func (worker *Worker) generate(ctx context.Context, tab *chatTab, turn GenerateT return &ProtectedResponse{ StatusCode: status, Header: header, + Proof: proof, Body: &protectedResponseBody{ ctx: bodyCtx, cancel: cancelBody, worker: worker, contextID: tab.contextID, page: tab.page, requestID: requestID, channel: channel, stop: func() error { return worker.stopTurn(tab) }, release: func(done bool) { worker.releaseChat(tab, done && status == http.StatusOK) }, @@ -357,44 +358,44 @@ func (worker *Worker) waitRunnable(ctx context.Context, tab *chatTab) error { } } -// waitSent 等页面应用发出本轮 GenerateContent；页面忙于处理或已开始运行时随之等待，点击后页面空闲时持续没有开始运行或弹出对话框时以页面给出的原因失败 -func (worker *Worker) waitSent(ctx context.Context, tab *chatTab, requestID string, channel *bidiChannel) error { +// waitSent 等页面应用发出本轮 GenerateContent 并返回页面写入的 proof；页面忙于处理或已开始运行时随之等待，点击后页面空闲时持续没有开始运行或弹出对话框时以页面给出的原因失败 +func (worker *Worker) waitSent(ctx context.Context, tab *chatTab, requestID string, channel *bidiChannel) (string, error) { for checks := 1; ; checks++ { pollCtx, cancel := context.WithTimeout(ctx, pagePollInterval) message, err := readProtectedMessage(pollCtx, channel) cancel() if err != nil && (ctx.Err() != nil || !errors.Is(err, context.DeadlineExceeded)) { - return err + return "", err } if err == nil { - return sentError(message) + return sentProof(message) } state, err := readRunState(ctx, worker.client, tab.contextID, tab.page, requestID) if err != nil { - return err + return "", err } if state.Started || state.Running { message, err := readProtectedMessage(ctx, channel) if err != nil { - return err + return "", err } - return sentError(message) + return sentProof(message) } if len(state.Dialogs) > 0 || checks == runSettleChecks { - return fmt.Errorf("点击 Run 后页面没有开始运行: %s", state.reason()) + return "", fmt.Errorf("点击 Run 后页面没有开始运行: %s", state.reason()) } } } -// sentError 返回通道上第一条消息表示的发送结果，页面已发出 GenerateContent 时为 nil -func sentError(message protectedMessage) error { +// sentProof 返回通道上第一条消息表示的发送结果，页面已发出 GenerateContent 时返回页面写入的 proof +func sentProof(message protectedMessage) (string, error) { if message.Error != "" { - return errors.New(message.Error) + return "", errors.New(message.Error) } if !message.Sent { - return errors.New("页面应用没有发出 GenerateContent") + return "", errors.New("页面应用没有发出 GenerateContent") } - return nil + return message.Proof, nil } // newChat 点击顶栏 New chat，在确认弹窗中丢弃当前对话，等页面清空并显示临时对话标识，临时对话中打开的新对话仍是临时对话 diff --git a/internal/camoufoxnative/protected.go b/internal/camoufoxnative/protected.go index 1308ec6a..f8f786a7 100644 --- a/internal/camoufoxnative/protected.go +++ b/internal/camoufoxnative/protected.go @@ -16,16 +16,18 @@ import ( "time" ) -// ProtectedResponse 表示由固定指纹浏览器流式返回的 MakerSuite 响应 +// ProtectedResponse 表示由固定指纹浏览器流式返回的 MakerSuite 响应，Proof 为界面运行时页面应用写入请求的 proof type ProtectedResponse struct { StatusCode int Header http.Header Body io.ReadCloser + Proof string } -// protectedMessage 是受保护请求通道上以 { 开头的控制消息：界面运行时先是发出标记，之后是状态码与响应头，最后是结束标记或错误；其余消息为 Base64 正文块 +// protectedMessage 是受保护请求通道上以 { 开头的控制消息：界面运行时先是带 proof 的发出标记，之后是状态码与响应头，最后是结束标记或错误；其余消息为 Base64 正文块 type protectedMessage struct { Sent bool `json:"sent"` + Proof string `json:"proof"` Status int `json:"status"` Headers [][2]string `json:"headers"` Done bool `json:"done"`