opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix codex streaming output and telegram duplicate retries

Browse source
This commit is contained in:
Glucksberg 2026-04-09 19:17:27 -04:00
parent a94ba82181
commit 5d127f149f
4 changed files with 166 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
pkg/channels/telegram/telegram.go
View file

@ -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)
}
}

25
pkg/channels/telegram/telegram_test.go
View file

@ -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) {

23
pkg/providers/oauth/codex_provider.go
View file

@ -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
}

93
pkg/providers/oauth/codex_provider_test.go
View file

@ -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))
}