From 5d127f149feb94240a79fcc7c85f08d927dc9ac9 Mon Sep 17 00:00:00 2001 From: Glucksberg <80581902+Glucksberg@users.noreply.github.com> Date: Thu, 9 Apr 2026 19:17:27 -0400 Subject: [PATCH] fix codex streaming output and telegram duplicate retries --- pkg/channels/telegram/telegram.go | 39 +++++++-- pkg/channels/telegram/telegram_test.go | 25 +++++- pkg/providers/oauth/codex_provider.go | 23 ++++++ pkg/providers/oauth/codex_provider_test.go | 93 ++++++++++++++++++++-- 4 files changed, 166 insertions(+), 14 deletions(-) diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index cebebfed6..e661cd6c4 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -350,12 +350,41 @@ func (c *TelegramChannel) sendChunk( pMsg, err := c.bot.SendMessage(ctx, tgMsg) if err != nil { - logParseFailed(err, params.useMarkdownV2) + // Only retry in plain text when Telegram explicitly rejected the formatted payload. + // Network/post-connect errors can mean the first message already landed, and retrying + // with different content creates visible duplicates in chat. + if strings.Contains(err.Error(), "Bad Request") { + logParseFailed(err, params.useMarkdownV2) - tgMsg.Text = params.mdFallback - tgMsg.ParseMode = "" - pMsg, err = c.bot.SendMessage(ctx, tgMsg) - if err != nil { + tgMsg.Text = params.mdFallback + tgMsg.ParseMode = "" + pMsg, err = c.bot.SendMessage(ctx, tgMsg) + if err != nil { + if isPostConnectError(err) { + logger.WarnCF( + "telegram", + "Send fallback likely landed but result is unknown; swallowing error to prevent duplicate", + map[string]any{ + "chat_id": params.chatID, + "error": err.Error(), + }, + ) + return "", nil + } + return "", fmt.Errorf("telegram send: %w", channels.ErrTemporary) + } + } else { + if isPostConnectError(err) { + logger.WarnCF( + "telegram", + "Send likely landed but result is unknown; swallowing error to prevent duplicate", + map[string]any{ + "chat_id": params.chatID, + "error": err.Error(), + }, + ) + return "", nil + } return "", fmt.Errorf("telegram send: %w", channels.ErrTemporary) } } diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index 69c76b430..19f3a3c8a 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -503,7 +503,7 @@ func TestSend_HTMLFallback_BothFail(t *testing.T) { assert.Error(t, err) assert.True(t, errors.Is(err, channels.ErrTemporary), "error should wrap ErrTemporary") - assert.Equal(t, 2, len(caller.calls), "should have HTML attempt + plain text attempt") + assert.Equal(t, 1, len(caller.calls), "should not retry as plain text for non-parse failures") } func TestSend_LongMessage_HTMLFallback_StopsOnError(t *testing.T) { @@ -524,8 +524,27 @@ func TestSend_LongMessage_HTMLFallback_StopsOnError(t *testing.T) { }) assert.Error(t, err) - // Should fail on the first chunk (2 calls: HTML + fallback), never reaching the second chunk. - assert.Equal(t, 2, len(caller.calls), "should stop after first chunk fails both HTML and plain text") + // Should fail on the first chunk (1 call: formatted send only), never reaching the second chunk. + assert.Equal(t, 1, len(caller.calls), "should stop after first chunk fails without plain-text retry") +} + +func TestSend_PostConnectError_SwallowsToPreventDuplicate(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return nil, errors.New("write tcp 10.0.0.1:12345->149.154.167.220:443: write: broken pipe") + }, + } + ch := newTestChannel(t, caller) + + msgIDs, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "12345", + Content: "Hello", + }) + + assert.NoError(t, err) + assert.Len(t, caller.calls, 1, "post-connect failure should not trigger plain-text retry") + assert.Len(t, msgIDs, 1, "channel should still report success to suppress duplicate retries") + assert.Equal(t, "", msgIDs[0], "message id is unknown when send result is swallowed") } func TestSend_MarkdownShortButHTMLLong_MultipleCalls(t *testing.T) { diff --git a/pkg/providers/oauth/codex_provider.go b/pkg/providers/oauth/codex_provider.go index 0b125997b..26d031fa7 100644 --- a/pkg/providers/oauth/codex_provider.go +++ b/pkg/providers/oauth/codex_provider.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sort" "strings" "github.com/openai/openai-go/v3" @@ -104,10 +105,18 @@ func (p *CodexProvider) Chat( defer stream.Close() var resp *responses.Response + streamOutput := map[int64]responses.ResponseOutputItemUnion{} for stream.Next() { evt := stream.Current() + switch evt.Type { + case "response.output_item.done": + streamOutput[evt.OutputIndex] = evt.Item + } if evt.Type == "response.completed" || evt.Type == "response.failed" || evt.Type == "response.incomplete" { evtResp := evt.Response + if len(evtResp.Output) == 0 && len(streamOutput) > 0 { + evtResp.Output = orderedStreamOutput(streamOutput) + } if evtResp.ID != "" { evtRespCopy := evtResp resp = &evtRespCopy @@ -156,6 +165,20 @@ func (p *CodexProvider) Chat( return orc.ParseResponseFromStruct(resp), nil } +func orderedStreamOutput(items map[int64]responses.ResponseOutputItemUnion) []responses.ResponseOutputItemUnion { + indexes := make([]int64, 0, len(items)) + for idx := range items { + indexes = append(indexes, idx) + } + sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] }) + + out := make([]responses.ResponseOutputItemUnion, 0, len(indexes)) + for _, idx := range indexes { + out = append(out, items[idx]) + } + return out +} + func (p *CodexProvider) GetDefaultModel() string { return codexDefaultModel } diff --git a/pkg/providers/oauth/codex_provider_test.go b/pkg/providers/oauth/codex_provider_test.go index aeeb18360..be248a16d 100644 --- a/pkg/providers/oauth/codex_provider_test.go +++ b/pkg/providers/oauth/codex_provider_test.go @@ -374,6 +374,84 @@ func TestCodexProvider_ChatRoundTrip(t *testing.T) { } } +func TestCodexProvider_ChatRoundTrip_UsesOutputItemDoneWhenCompletedOutputIsEmpty(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/responses" { + http.Error(w, "not found: "+r.URL.Path, http.StatusNotFound) + return + } + if r.Header.Get("Authorization") != "Bearer test-token" { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if r.Header.Get("Chatgpt-Account-Id") != "acc-123" { + http.Error(w, "missing account id", http.StatusBadRequest) + return + } + + var reqBody map[string]any + if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { + http.Error(w, "invalid json", http.StatusBadRequest) + return + } + if reqBody["stream"] != true { + http.Error(w, "stream must be true", http.StatusBadRequest) + return + } + + outputItem := map[string]any{ + "id": "msg_1", + "type": "message", + "role": "assistant", + "status": "completed", + "content": []map[string]any{ + {"type": "output_text", "text": "OK"}, + }, + } + completed := map[string]any{ + "id": "resp_test", + "object": "response", + "status": "completed", + "output": []map[string]any{}, + "usage": map[string]any{ + "input_tokens": 23, + "output_tokens": 5, + "total_tokens": 28, + "input_tokens_details": map[string]any{"cached_tokens": 0}, + "output_tokens_details": map[string]any{"reasoning_tokens": 0}, + }, + } + + writeSSEEvent(w, "response.output_item.done", map[string]any{ + "type": "response.output_item.done", + "sequence_number": 7, + "output_index": 0, + "item": outputItem, + }) + writeSSEEvent(w, "response.completed", map[string]any{ + "type": "response.completed", + "sequence_number": 8, + "response": completed, + }) + fmt.Fprintf(w, "data: [DONE]\n\n") + })) + defer server.Close() + + provider := NewCodexProvider("test-token", "acc-123") + provider.client = createOpenAITestClient(server.URL, "test-token", "acc-123") + + resp, err := provider.Chat(t.Context(), []Message{{Role: "user", Content: "Hello"}}, nil, "gpt-5.4", nil) + if err != nil { + t.Fatalf("Chat() error: %v", err) + } + if resp.Content != "OK" { + t.Fatalf("Content = %q, want %q", resp.Content, "OK") + } + if resp.Usage == nil || resp.Usage.TotalTokens != 28 { + t.Fatalf("Usage.TotalTokens = %#v, want 28", resp.Usage) + } +} + func TestCodexProvider_ChatRoundTrip_WebSearchDisabled(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/responses" { @@ -636,14 +714,17 @@ func createOpenAITestClient(baseURL, token, accountID string) *openai.Client { } func writeCompletedSSE(w http.ResponseWriter, response map[string]any) { - event := map[string]any{ + w.Header().Set("Content-Type", "text/event-stream") + writeSSEEvent(w, "response.completed", map[string]any{ "type": "response.completed", "sequence_number": 1, "response": response, - } - b, _ := json.Marshal(event) - w.Header().Set("Content-Type", "text/event-stream") - fmt.Fprintf(w, "event: response.completed\n") - fmt.Fprintf(w, "data: %s\n\n", string(b)) + }) fmt.Fprintf(w, "data: [DONE]\n\n") } + +func writeSSEEvent(w http.ResponseWriter, eventName string, payload map[string]any) { + b, _ := json.Marshal(payload) + fmt.Fprintf(w, "event: %s\n", eventName) + fmt.Fprintf(w, "data: %s\n\n", string(b)) +}