diff --git a/backend/internal/provider/chatgpt/client.go b/backend/internal/provider/chatgpt/client.go index b78d2e6..a6fc18b 100644 --- a/backend/internal/provider/chatgpt/client.go +++ b/backend/internal/provider/chatgpt/client.go @@ -898,29 +898,41 @@ func (c *Client) startImageGeneration(ctx context.Context, session tlsclient.Htt scanner := bufio.NewScanner(resp.Body) scanner.Buffer(make([]byte, 0, 1024*1024), 8*1024*1024) var chunks []string - sseStart := time.Now() - // Watchdog: once the conversation id is known, the stream may go silent - // without ever emitting the async marker, leaving scanner.Scan() blocked on - // a read for the whole ctx budget. Closing the body unblocks the read so the - // loop exits and we fall through to polling. - convFound := make(chan struct{}) + refIDs := uploadedRefIDSet(refs) + // Watchdog: scanner.Scan() blocks on a read while the SSE is silent, which + // would hold the whole ctx budget. ChatGPT's inline image pipeline keeps the + // connection open with periodic ": ping" keepalives (~15s apart) while the + // image renders, then streams the asset (a role:"tool" multimodal_text turn + // carrying image_asset_pointer / sediment://) in the same SSE. So this is an + // *idle* timeout reset on every received line — a fixed grace from the + // conversation id would abort mid-render before the asset arrives. It only + // closes the body (unblocking the read) once the stream is genuinely silent. + activity := make(chan struct{}, 1) watchdogDone := make(chan struct{}) defer close(watchdogDone) go func() { - select { - case <-convFound: - case <-watchdogDone: - return - } - timer := time.NewTimer(sseAsyncGrace) + timer := time.NewTimer(sseIdleGrace) defer timer.Stop() - select { - case <-timer.C: - resp.Body.Close() - case <-watchdogDone: + for { + select { + case <-activity: + if !timer.Stop() { + <-timer.C + } + timer.Reset(sseIdleGrace) + case <-timer.C: + resp.Body.Close() + return + case <-watchdogDone: + return + } } }() for scanner.Scan() { + select { + case activity <- struct{}{}: + default: + } line := scanner.Text() if !strings.HasPrefix(line, "data:") { continue @@ -936,7 +948,6 @@ func (c *Client) startImageGeneration(ctx context.Context, session tlsclient.Htt if conversationID == "" { if match := conversationIDRE.FindStringSubmatch(payload); len(match) >= 2 { conversationID = match[1] - close(convFound) } } newFiles, newSeds := scanForIDs(payload) @@ -945,18 +956,15 @@ func (c *Client) startImageGeneration(ctx context.Context, session tlsclient.Htt if !asyncStarted && containsAsyncMarker(payload) { asyncStarted = true } - // Async pipeline: ChatGPT no longer streams the image inline — it returns - // a placeholder tool turn (image_gen_async / image_gen_task_id) and - // delivers the asset later via conversation polling. Once we have the - // conversation id there is nothing more to read here, so stop instead of - // holding the SSE open until [DONE] (a stalled stream would otherwise burn - // the whole generation budget and surface as "context deadline exceeded"). - // - // The async marker normally arrives within ~1s of the conversation id. We - // still only wait a short grace for it (the watchdog above unblocks a - // silent stream) so a request that never engages the async pipeline is - // detected quickly instead of burning the whole budget. - if conversationID != "" && (asyncStarted || time.Since(sseStart) >= sseAsyncGrace) { + // Stop once the image pipeline is confirmed engaged (async/tool marker) or + // the generated asset id has streamed inline. The stream echoes back the + // user's uploaded reference images too, so exclude those ids before deciding + // — otherwise an edit request would stop on its own upload. The idle watchdog + // above bounds a silent stream; pollForImage backstops asset retrieval when + // nothing streamed inline. + haveAsset := len(dropIDs(append([]string(nil), fileIDs...), refIDs)) > 0 || + len(dropIDs(append([]string(nil), sedimentIDs...), refIDs)) > 0 + if conversationID != "" && (asyncStarted || haveAsset) { break } } @@ -969,15 +977,13 @@ func (c *Client) startImageGeneration(ctx context.Context, session tlsclient.Htt if conversationID == "" { return "", nil, nil, errors.New("chatgpt SSE closed without conversation_id") } - // Intermittently (~10% on gpt-5-5-thinking) the stream returns a conversation - // id but never emits the async pipeline marker and no image is ever produced — - // polling such a conversation only burns the whole budget and surfaces as the - // non-retryable "image poll timeout". The async marker is the reliable "the - // image generation task actually started" signal, so when it is absent (and - // nothing was streamed inline) treat the attempt as a transient upstream - // failure. That is retryable: a fresh submission reliably engages the pipeline, - // so the pool retries the same account a few times and then fails over to - // another account (换号重试) instead of failing the request. + // The stream closed with a conversation id but no sign the image pipeline + // engaged — no async/tool marker and no asset streamed inline. Polling such a + // conversation only burns the whole budget and surfaces as the non-retryable + // "image poll timeout", so treat the attempt as a transient upstream failure. + // That is retryable: a fresh submission reliably engages the pipeline, so the + // pool retries the same account a few times and then fails over to another + // account (换号重试) instead of failing the request. if !asyncStarted && len(fileIDs) == 0 && len(sedimentIDs) == 0 { return "", nil, nil, fmt.Errorf("%w: image generation did not start (no async marker)", ErrTemporaryUpstream) } diff --git a/backend/internal/provider/chatgpt/util.go b/backend/internal/provider/chatgpt/util.go index 5243c52..4ac2fd0 100644 --- a/backend/internal/provider/chatgpt/util.go +++ b/backend/internal/provider/chatgpt/util.go @@ -18,12 +18,13 @@ const ( defaultClientVersion = "prod-db390ebea64862bf1899c420a4c736e0cf639747" defaultClientBuildNumber = "7904904" defaultPOWScript = "https://chatgpt.com/backend-api/sentinel/sdk.js" - // sseAsyncGrace bounds how long startImageGeneration keeps reading the SSE - // waiting for the image_gen_async marker after the conversation id is known. - // The marker normally arrives within ~1s; when it never streams (intermittent - // on gpt-5-5-thinking) we must not hold the stream open for the whole ctx - // budget, so we break after this grace and fall through to polling. - sseAsyncGrace = 10 * time.Second + // sseIdleGrace bounds how long startImageGeneration keeps reading the image + // SSE with no activity at all before giving up. ChatGPT's inline image pipeline + // holds the stream open with periodic ": ping" keepalives (~15s apart) while + // the image renders, so this is an *idle* timeout reset on every received line + // (see the watchdog in startImageGeneration): it only fires when the stream is + // genuinely silent, not on the normal render wait. + sseIdleGrace = 45 * time.Second // pictureV2Command is the "@创建图片" ecosystem mention the ChatGPT web app // prepends to an image (picture_v2) prompt; the server identifies the image @@ -56,7 +57,7 @@ var ( // async image pipeline (image is delivered later via conversation polling // rather than inline in the SSE stream). Their presence means "generating — // keep polling", NOT failure. - asyncMarkers = []string{"image_gen_async", "image_gen_task_id", "trigger_async_ux"} + asyncMarkers = []string{"image_gen_async", "image_gen_task_id", "trigger_async_ux", "ImageGenToolTemporal", "image_gen_title"} // contentPolicyMarkers are stable substrings of ChatGPT's content-audit // refusal message. When one appears in an assistant turn the prompt was