From 02f1c12728eae9e49cf9346ea32fd18cab4d32e8 Mon Sep 17 00:00:00 2001 From: lxowalle Date: Tue, 21 Apr 2026 17:18:12 +0800 Subject: [PATCH] fix review blockers in pico token cache and tool feedback fix(provider): preserve function thought signatures fix(feishu): recover tool feedback after edit fallback --- pkg/agent/hooks_test.go | 88 +++++++++++++++++ pkg/agent/pipeline_execute.go | 2 +- pkg/agent/subturn_test.go | 32 +++++++ pkg/agent/turn_state.go | 6 +- pkg/channels/feishu/feishu_64.go | 37 ++++++- pkg/channels/feishu/feishu_64_test.go | 26 +++++ pkg/channels/manager.go | 37 +++++-- pkg/channels/manager_test.go | 105 ++++++++++++++++++++ pkg/channels/telegram/telegram.go | 52 +++++++--- pkg/channels/telegram/telegram_test.go | 127 +++++++++++++++++++++++++ pkg/providers/common/common.go | 18 ++-- pkg/providers/common/common_test.go | 24 +++++ web/backend/api/pico.go | 4 + web/backend/api/pico_test.go | 52 ++++++++++ 14 files changed, 575 insertions(+), 35 deletions(-) diff --git a/pkg/agent/hooks_test.go b/pkg/agent/hooks_test.go index cd1586e75..b31af5582 100644 --- a/pkg/agent/hooks_test.go +++ b/pkg/agent/hooks_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "os" + "strings" "sync" "testing" "time" @@ -403,6 +404,24 @@ func (h *toolRewriteHook) AfterTool( return next, HookDecision{Action: HookActionModify}, nil } +type toolRenameHook struct{} + +func (h *toolRenameHook) BeforeTool( + ctx context.Context, + call *ToolCallHookRequest, +) (*ToolCallHookRequest, HookDecision, error) { + next := call.Clone() + next.Tool = "echo_text_rewritten" + return next, HookDecision{Action: HookActionModify}, nil +} + +func (h *toolRenameHook) AfterTool( + ctx context.Context, + result *ToolResultHookResponse, +) (*ToolResultHookResponse, HookDecision, error) { + return result.Clone(), HookDecision{Action: HookActionContinue}, nil +} + func TestAgentLoop_Hooks_ToolInterceptorCanRewrite(t *testing.T) { provider := &toolHookProvider{} al, agent, cleanup := newHookTestLoop(t, provider) @@ -430,6 +449,75 @@ func TestAgentLoop_Hooks_ToolInterceptorCanRewrite(t *testing.T) { } } +type echoTextRewrittenTool struct{} + +func (t *echoTextRewrittenTool) Name() string { + return "echo_text_rewritten" +} + +func (t *echoTextRewrittenTool) Description() string { + return "echo a rewritten text argument" +} + +func (t *echoTextRewrittenTool) Parameters() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "text": map[string]any{ + "type": "string", + }, + }, + } +} + +func (t *echoTextRewrittenTool) Execute(ctx context.Context, args map[string]any) *tools.ToolResult { + text, _ := args["text"].(string) + return tools.SilentResult("rewritten:" + text) +} + +func TestAgentLoop_Hooks_ToolFeedbackUsesRewrittenToolName(t *testing.T) { + provider := &toolHookProvider{} + al, agent, cleanup := newHookTestLoop(t, provider) + defer cleanup() + + al.cfg.Agents.Defaults.ToolFeedback.Enabled = true + al.RegisterTool(&echoTextTool{}) + al.RegisterTool(&echoTextRewrittenTool{}) + if err := al.MountHook(NamedHook("tool-rename", &toolRenameHook{})); err != nil { + t.Fatalf("MountHook failed: %v", err) + } + + _, err := al.runAgentLoop(context.Background(), agent, processOptions{ + SessionKey: "session-1", + Channel: "cli", + ChatID: "direct", + UserMessage: "run tool", + DefaultResponse: defaultResponse, + EnableSummary: false, + SendResponse: false, + }) + if err != nil { + t.Fatalf("runAgentLoop failed: %v", err) + } + + msgBus, ok := al.bus.(*bus.MessageBus) + if !ok { + t.Fatalf("expected concrete MessageBus, got %T", al.bus) + } + + select { + case outbound := <-msgBus.OutboundChan(): + if !strings.Contains(outbound.Content, "`echo_text_rewritten`") { + t.Fatalf("tool feedback content = %q, want rewritten tool name", outbound.Content) + } + if strings.Contains(outbound.Content, "`echo_text`") { + t.Fatalf("tool feedback content = %q, want no original tool name", outbound.Content) + } + case <-time.After(2 * time.Second): + t.Fatal("expected outbound tool feedback") + } +} + type denyApprovalHook struct{} func (h *denyApprovalHook) ApproveTool(ctx context.Context, req *ToolApprovalRequest) (ApprovalDecision, error) { diff --git a/pkg/agent/pipeline_execute.go b/pkg/agent/pipeline_execute.go index 8acd32774..c19149911 100644 --- a/pkg/agent/pipeline_execute.go +++ b/pkg/agent/pipeline_execute.go @@ -349,7 +349,7 @@ toolLoop: messages, al.cfg.Agents.Defaults.GetToolFeedbackMaxArgsLength(), ) - feedbackMsg := utils.FormatToolFeedbackMessage(tc.Name, toolFeedbackExplanation) + feedbackMsg := utils.FormatToolFeedbackMessage(toolName, toolFeedbackExplanation) fbCtx, fbCancel := context.WithTimeout(turnCtx, 3*time.Second) _ = al.bus.PublishOutbound(fbCtx, outboundMessageForTurnWithKind(ts, feedbackMsg, messageKindToolFeedback)) fbCancel() diff --git a/pkg/agent/subturn_test.go b/pkg/agent/subturn_test.go index 6a2ba835d..040063249 100644 --- a/pkg/agent/subturn_test.go +++ b/pkg/agent/subturn_test.go @@ -1650,6 +1650,38 @@ func TestGrandchildAbort_CascadingCancellation(t *testing.T) { } } +func TestNestedSubTurn_GracefulFinishSignalsDirectChildren(t *testing.T) { + parentCtx := context.Background() + parentTS := &turnState{ + ctx: parentCtx, + turnID: "parent-graceful", + depth: 1, + pendingResults: make(chan *tools.ToolResult, 16), + } + parentTS.ctx, parentTS.cancelFunc = context.WithCancel(parentCtx) + + childTS := &turnState{ + ctx: context.Background(), + turnID: "child-graceful", + depth: 2, + parentTurnState: parentTS, + pendingResults: make(chan *tools.ToolResult, 16), + } + + if childTS.IsParentEnded() { + t.Fatal("IsParentEnded should be false before parent finishes") + } + + parentTS.Finish(false) + + if !parentTS.parentEnded.Load() { + t.Fatal("parentEnded should be true after graceful finish") + } + if !childTS.IsParentEnded() { + t.Fatal("nested child should observe parent graceful finish") + } +} + // TestSpawnDuringAbort_RaceCondition verifies behavior when trying to spawn // a sub-turn while the parent is being aborted. func TestSpawnDuringAbort_RaceCondition(t *testing.T) { diff --git a/pkg/agent/turn_state.go b/pkg/agent/turn_state.go index edf8654b5..8b5fd4e2c 100644 --- a/pkg/agent/turn_state.go +++ b/pkg/agent/turn_state.go @@ -554,9 +554,9 @@ func (ts *turnState) Finish(isHardAbort bool) { ts.mu.Unlock() }) - // If this is a graceful finish (not hard abort), signal to children - if !isHardAbort && ts.parentTurnState == nil { - // This is a root turn finishing gracefully + // Any graceful finish must signal direct children so nested SubTurns can + // observe parent completion and decide whether to stop or continue. + if !isHardAbort { ts.parentEnded.Store(true) } diff --git a/pkg/channels/feishu/feishu_64.go b/pkg/channels/feishu/feishu_64.go index 49b8dd8e5..8f3ae39d9 100644 --- a/pkg/channels/feishu/feishu_64.go +++ b/pkg/channels/feishu/feishu_64.go @@ -50,7 +50,8 @@ type FeishuChannel struct { mu sync.Mutex cancel context.CancelFunc - progress *channels.ToolFeedbackAnimator + progress *channels.ToolFeedbackAnimator + deleteMessageFn func(context.Context, string, string) error } type cachedMessage struct { @@ -76,6 +77,7 @@ func NewFeishuChannel(bc *config.Channel, cfg *config.FeishuSettings, bus *bus.M tokenCache: tc, client: lark.NewClient(cfg.AppID, cfg.AppSecret.String(), opts...), } + ch.deleteMessageFn = ch.deleteMessageAPI ch.progress = channels.NewToolFeedbackAnimator(ch.EditMessage) ch.SetOwner(ch) return ch, nil @@ -156,19 +158,24 @@ func (c *FeishuChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]st } isToolFeedback := outboundMessageIsToolFeedback(msg) - trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(msg.ChatID) if isToolFeedback { if msgID, handled, err := c.progress.Update(ctx, msg.ChatID, msg.Content); handled { if err != nil { - return nil, err + // Feishu can fall back to plain text for a previous progress + // message, and those messages cannot be patched through the card + // edit API. Drop the stale tracker and recreate the progress + // message so later tool feedback is not blocked. + c.resetTrackedToolFeedbackAfterEditFailure(ctx, msg.ChatID) + } else { + return []string{msgID}, nil } - return []string{msgID}, nil } } else { if msgIDs, handled := c.FinalizeToolFeedbackMessage(ctx, msg); handled { return msgIDs, nil } } + trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(msg.ChatID) // Build interactive card with markdown content sendContent := msg.Content @@ -256,6 +263,14 @@ func (c *FeishuChannel) EditMessage(ctx context.Context, chatID, messageID, cont // DeleteMessage implements channels.MessageDeleter. func (c *FeishuChannel) DeleteMessage(ctx context.Context, chatID, messageID string) error { + deleteFn := c.deleteMessageFn + if deleteFn == nil { + deleteFn = c.deleteMessageAPI + } + return deleteFn(ctx, chatID, messageID) +} + +func (c *FeishuChannel) deleteMessageAPI(ctx context.Context, chatID, messageID string) error { req := larkim.NewDeleteMessageReqBuilder(). MessageId(messageID). Build() @@ -355,12 +370,24 @@ func (c *FeishuChannel) DismissToolFeedbackMessage(ctx context.Context, chatID s c.dismissTrackedToolFeedbackMessage(ctx, chatID, msgID) } +func (c *FeishuChannel) resetTrackedToolFeedbackAfterEditFailure(ctx context.Context, chatID string) { + msgID, ok := c.currentToolFeedbackMessage(chatID) + if !ok { + return + } + c.dismissTrackedToolFeedbackMessage(ctx, chatID, msgID) +} + func (c *FeishuChannel) dismissTrackedToolFeedbackMessage(ctx context.Context, chatID, messageID string) { if strings.TrimSpace(chatID) == "" || strings.TrimSpace(messageID) == "" { return } c.ClearToolFeedbackMessage(chatID) - _ = c.DeleteMessage(ctx, chatID, messageID) + deleteFn := c.deleteMessageFn + if deleteFn == nil { + deleteFn = c.deleteMessageAPI + } + _ = deleteFn(ctx, chatID, messageID) } func (c *FeishuChannel) finalizeTrackedToolFeedbackMessage( diff --git a/pkg/channels/feishu/feishu_64_test.go b/pkg/channels/feishu/feishu_64_test.go index 0bdac0352..48fdf0f74 100644 --- a/pkg/channels/feishu/feishu_64_test.go +++ b/pkg/channels/feishu/feishu_64_test.go @@ -364,3 +364,29 @@ func TestFinalizeTrackedToolFeedbackMessage_EditFailureKeepsTrackedMessage(t *te t.Fatalf("expected tracked tool feedback to remain after failed edit, got (%q, %v)", msgID, ok) } } + +func TestResetTrackedToolFeedbackAfterEditFailure_DismissesTrackedMessage(t *testing.T) { + var ( + deletedChatID string + deletedMsgID string + ) + + ch := &FeishuChannel{ + progress: channels.NewToolFeedbackAnimator(nil), + deleteMessageFn: func(_ context.Context, chatID, messageID string) error { + deletedChatID = chatID + deletedMsgID = messageID + return nil + }, + } + ch.RecordToolFeedbackMessage("chat-1", "msg-1", "🔧 `read_file`") + + ch.resetTrackedToolFeedbackAfterEditFailure(context.Background(), "chat-1") + + if deletedChatID != "chat-1" || deletedMsgID != "msg-1" { + t.Fatalf("unexpected delete target: chat=%q msg=%q", deletedChatID, deletedMsgID) + } + if _, ok := ch.currentToolFeedbackMessage("chat-1"); ok { + t.Fatal("expected tracked tool feedback to be cleared after edit failure reset") + } +} diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 6aec966d6..b2df35414 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -107,6 +107,10 @@ type toolFeedbackMessageCleaner interface { DismissToolFeedbackMessage(ctx context.Context, chatID string) } +type toolFeedbackMessageTargetResolver interface { + ToolFeedbackMessageChatID(chatID string, outboundCtx *bus.InboundContext) string +} + type asyncTask struct { cancel context.CancelFunc } @@ -134,13 +138,31 @@ func outboundMediaChatID(msg bus.OutboundMediaMessage) string { return msg.ChatID } -func dismissTrackedToolFeedbackMessage(ctx context.Context, ch Channel, chatID string) { +func trackedToolFeedbackMessageChatID(ch Channel, chatID string, outboundCtx *bus.InboundContext) string { + if resolver, ok := ch.(toolFeedbackMessageTargetResolver); ok { + if resolved := strings.TrimSpace(resolver.ToolFeedbackMessageChatID(chatID, outboundCtx)); resolved != "" { + return resolved + } + } + return strings.TrimSpace(chatID) +} + +func dismissTrackedToolFeedbackMessage( + ctx context.Context, + ch Channel, + chatID string, + outboundCtx *bus.InboundContext, +) { + trackedChatID := trackedToolFeedbackMessageChatID(ch, chatID, outboundCtx) + if trackedChatID == "" { + return + } if cleaner, ok := ch.(toolFeedbackMessageCleaner); ok { - cleaner.DismissToolFeedbackMessage(ctx, chatID) + cleaner.DismissToolFeedbackMessage(ctx, trackedChatID) return } if tracker, ok := ch.(toolFeedbackMessageTracker); ok { - tracker.ClearToolFeedbackMessage(chatID) + tracker.ClearToolFeedbackMessage(trackedChatID) } } @@ -249,7 +271,7 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess } } if !isToolFeedback { - dismissTrackedToolFeedbackMessage(ctx, ch, chatID) + dismissTrackedToolFeedbackMessage(ctx, ch, chatID, &msg.Context) } return nil, true } @@ -263,10 +285,11 @@ func (m *Manager) preSend(ctx context.Context, name string, msg bus.OutboundMess content = InitialAnimatedToolFeedbackContent(msg.Content) } if err := editor.EditMessage(ctx, chatID, entry.id, content); err == nil { + trackedChatID := trackedToolFeedbackMessageChatID(ch, chatID, &msg.Context) if tracker, ok := ch.(toolFeedbackMessageTracker); ok && isToolFeedback { - tracker.RecordToolFeedbackMessage(chatID, entry.id, msg.Content) + tracker.RecordToolFeedbackMessage(trackedChatID, entry.id, msg.Content) } else if !isToolFeedback { - dismissTrackedToolFeedbackMessage(ctx, ch, chatID) + dismissTrackedToolFeedbackMessage(ctx, ch, chatID, &msg.Context) } return []string{entry.id}, true } @@ -366,7 +389,7 @@ func (m *Manager) GetStreamer(ctx context.Context, channelName, chatID string) ( return &finalizeHookStreamer{ Streamer: streamer, onFinalize: func(finalizeCtx context.Context) { - dismissTrackedToolFeedbackMessage(finalizeCtx, ch, chatID) + dismissTrackedToolFeedbackMessage(finalizeCtx, ch, chatID, nil) m.streamActive.Store(key, true) }, }, true diff --git a/pkg/channels/manager_test.go b/pkg/channels/manager_test.go index 4f6a7dcf4..31e77704d 100644 --- a/pkg/channels/manager_test.go +++ b/pkg/channels/manager_test.go @@ -786,6 +786,21 @@ func (m *mockMessageEditor) FinalizeToolFeedbackMessage( return m.finalizeFn(ctx, msg) } +type mockResolvedToolFeedbackEditor struct { + mockMessageEditor + resolveChatIDFn func(chatID string, outboundCtx *bus.InboundContext) string +} + +func (m *mockResolvedToolFeedbackEditor) ToolFeedbackMessageChatID( + chatID string, + outboundCtx *bus.InboundContext, +) string { + if m.resolveChatIDFn != nil { + return m.resolveChatIDFn(chatID, outboundCtx) + } + return chatID +} + func TestPreSend_PlaceholderEditSuccess(t *testing.T) { m := newTestManager() var sendCalled bool @@ -865,6 +880,54 @@ func TestPreSend_ToolFeedbackPlaceholderEditRecordsTrackedMessage(t *testing.T) } } +func TestPreSend_ToolFeedbackPlaceholderEditUsesResolvedTrackedChatID(t *testing.T) { + m := newTestManager() + + ch := &mockResolvedToolFeedbackEditor{ + mockMessageEditor: mockMessageEditor{ + editFn: func(_ context.Context, chatID, messageID, content string) error { + if chatID != "-100123" || messageID != "456" || content != "hello" { + t.Fatalf("unexpected edit args: %s %s %s", chatID, messageID, content) + } + return nil + }, + }, + resolveChatIDFn: func(chatID string, outboundCtx *bus.InboundContext) string { + if chatID != "-100123" { + t.Fatalf("expected raw chat ID, got %q", chatID) + } + if outboundCtx == nil || outboundCtx.TopicID != "42" { + t.Fatalf("expected topic-aware outbound context, got %+v", outboundCtx) + } + return chatID + "/" + outboundCtx.TopicID + }, + } + + m.RecordPlaceholder("test", "-100123", "456") + + msg := testOutboundMessage(bus.OutboundMessage{ + Channel: "test", + ChatID: "-100123", + Content: "hello", + Context: bus.InboundContext{ + Channel: "test", + ChatID: "-100123", + TopicID: "42", + Raw: map[string]string{ + "message_kind": "tool_feedback", + }, + }, + }) + _, edited := m.preSend(context.Background(), "test", msg, ch) + if !edited { + t.Fatal("expected preSend to edit placeholder") + } + if ch.recordedChatID != "-100123/42" || ch.recordedMessageID != "456" { + t.Fatalf("expected resolved tracked message -100123/42/456, got %q/%q", + ch.recordedChatID, ch.recordedMessageID) + } +} + func TestPreSend_NonToolFeedbackLeavesTrackedMessageForChannelSend(t *testing.T) { m := newTestManager() ch := &mockMessageEditor{} @@ -1094,6 +1157,48 @@ func TestGetStreamer_FinalizeDismissesTrackedToolFeedback(t *testing.T) { } } +func TestPreSend_PlaceholderEditSuccessDismissesResolvedTrackedToolFeedback(t *testing.T) { + m := newTestManager() + + ch := &mockResolvedToolFeedbackEditor{ + mockMessageEditor: mockMessageEditor{ + editFn: func(_ context.Context, chatID, messageID, content string) error { + if chatID != "-100123" || messageID != "456" || content != "done" { + t.Fatalf("unexpected edit args: %s %s %s", chatID, messageID, content) + } + return nil + }, + }, + resolveChatIDFn: func(chatID string, outboundCtx *bus.InboundContext) string { + if outboundCtx == nil || outboundCtx.TopicID != "42" { + t.Fatalf("expected topic-aware outbound context, got %+v", outboundCtx) + } + return chatID + "/" + outboundCtx.TopicID + }, + } + + m.RecordPlaceholder("test", "-100123", "456") + + msg := testOutboundMessage(bus.OutboundMessage{ + Channel: "test", + ChatID: "-100123", + Content: "done", + Context: bus.InboundContext{ + Channel: "test", + ChatID: "-100123", + TopicID: "42", + }, + }) + + _, edited := m.preSend(context.Background(), "test", msg, ch) + if !edited { + t.Fatal("expected preSend to edit placeholder") + } + if ch.dismissedChatID != "-100123/42" { + t.Fatalf("expected resolved tracked dismissal, got %q", ch.dismissedChatID) + } +} + func TestGetStreamer_FinalizeFailureDoesNotDismissTrackedToolFeedback(t *testing.T) { m := newTestManager() ch := &mockStreamingChannel{ diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 8bec7856d..cc148fd5b 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -203,17 +203,18 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] if isToolFeedback { toolFeedbackContent = fitToolFeedbackForTelegram(msg.Content, useMarkdownV2, 4096) } + trackedChatID := telegramToolFeedbackChatKey(msg.ChatID, &msg.Context) if isToolFeedback { - if msgID, handled, err := c.progress.Update(ctx, msg.ChatID, toolFeedbackContent); handled { + if msgID, handled, err := c.progress.Update(ctx, trackedChatID, toolFeedbackContent); handled { if err != nil { return nil, err } return []string{msgID}, nil } } - trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(msg.ChatID) + trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(trackedChatID) if !isToolFeedback { - if msgIDs, handled := c.FinalizeToolFeedbackMessage(ctx, msg); handled { + if msgIDs, handled := c.finalizeToolFeedbackMessageForChat(ctx, trackedChatID, msg); handled { return msgIDs, nil } } @@ -308,9 +309,9 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([] } if isToolFeedback && len(messageIDs) > 0 { - c.RecordToolFeedbackMessage(msg.ChatID, messageIDs[0], toolFeedbackContent) + c.RecordToolFeedbackMessage(trackedChatID, messageIDs[0], toolFeedbackContent) } else if !isToolFeedback && hasTrackedMsg { - c.dismissTrackedToolFeedbackMessage(ctx, msg.ChatID, trackedMsgID) + c.dismissTrackedToolFeedbackMessage(ctx, trackedChatID, trackedMsgID) } return messageIDs, nil @@ -552,7 +553,15 @@ func (c *TelegramChannel) FinalizeToolFeedbackMessage(ctx context.Context, msg b if outboundMessageIsToolFeedback(msg) { return nil, false } - return c.finalizeTrackedToolFeedbackMessage(ctx, msg.ChatID, msg.Content, c.EditMessage) + return c.finalizeToolFeedbackMessageForChat(ctx, telegramToolFeedbackChatKey(msg.ChatID, &msg.Context), msg) +} + +func (c *TelegramChannel) finalizeToolFeedbackMessageForChat( + ctx context.Context, + chatID string, + msg bus.OutboundMessage, +) ([]string, bool) { + return c.finalizeTrackedToolFeedbackMessage(ctx, chatID, msg.Content, c.EditMessage) } // SendPlaceholder implements channels.PlaceholderCapable. @@ -586,7 +595,8 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe if !c.IsRunning() { return nil, channels.ErrNotRunning } - trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(msg.ChatID) + trackedChatID := telegramToolFeedbackChatKey(msg.ChatID, &msg.Context) + trackedMsgID, hasTrackedMsg := c.currentToolFeedbackMessage(trackedChatID) chatID, threadID, err := resolveTelegramOutboundTarget(msg.ChatID, &msg.Context) if err != nil { @@ -696,7 +706,7 @@ func (c *TelegramChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMe } if hasTrackedMsg { - c.dismissTrackedToolFeedbackMessage(ctx, msg.ChatID, trackedMsgID) + c.dismissTrackedToolFeedbackMessage(ctx, trackedChatID, trackedMsgID) } return messageIDs, nil @@ -1105,6 +1115,18 @@ func fitToolFeedbackForTelegram(content string, useMarkdownV2 bool, maxParsedLen return best } +func telegramToolFeedbackChatKey(chatID string, outboundCtx *bus.InboundContext) string { + resolvedChatID, threadID, err := resolveTelegramOutboundTarget(chatID, outboundCtx) + if err != nil || threadID == 0 { + return strings.TrimSpace(chatID) + } + return fmt.Sprintf("%d/%d", resolvedChatID, threadID) +} + +func (c *TelegramChannel) ToolFeedbackMessageChatID(chatID string, outboundCtx *bus.InboundContext) string { + return telegramToolFeedbackChatKey(chatID, outboundCtx) +} + // parseTelegramChatID splits "chatID/threadID" into its components. // Returns threadID=0 when no "/" is present (non-forum messages). func parseTelegramChatID(chatID string) (int64, int, error) { @@ -1255,7 +1277,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann return nil, fmt.Errorf("streaming disabled in config") } - cid, _, err := parseTelegramChatID(chatID) + cid, threadID, err := parseTelegramChatID(chatID) if err != nil { return nil, err } @@ -1264,6 +1286,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann return &telegramStreamer{ bot: c.bot, chatID: cid, + threadID: threadID, draftID: cryptoRandInt(), throttleInterval: time.Duration(streamCfg.ThrottleSeconds) * time.Second, minGrowth: streamCfg.MinGrowthChars, @@ -1276,6 +1299,7 @@ func (c *TelegramChannel) BeginStream(ctx context.Context, chatID string) (chann type telegramStreamer struct { bot *telego.Bot chatID int64 + threadID int draftID int throttleInterval time.Duration minGrowth int @@ -1303,10 +1327,11 @@ func (s *telegramStreamer) Update(ctx context.Context, content string) error { htmlContent := markdownToTelegramHTML(content) err := s.bot.SendMessageDraft(ctx, &telego.SendMessageDraftParams{ - ChatID: s.chatID, - DraftID: s.draftID, - Text: htmlContent, - ParseMode: telego.ModeHTML, + ChatID: s.chatID, + MessageThreadID: s.threadID, + DraftID: s.draftID, + Text: htmlContent, + ParseMode: telego.ModeHTML, }) if err != nil { // First error → degrade silently (e.g. no forum mode) @@ -1325,6 +1350,7 @@ func (s *telegramStreamer) Update(ctx context.Context, content string) error { func (s *telegramStreamer) Finalize(ctx context.Context, content string) error { htmlContent := markdownToTelegramHTML(content) tgMsg := tu.Message(tu.ID(s.chatID), htmlContent) + tgMsg.MessageThreadID = s.threadID tgMsg.ParseMode = telego.ModeHTML if _, err := s.bot.SendMessage(ctx, tgMsg); err != nil { diff --git a/pkg/channels/telegram/telegram_test.go b/pkg/channels/telegram/telegram_test.go index f3974723d..69c76b430 100644 --- a/pkg/channels/telegram/telegram_test.go +++ b/pkg/channels/telegram/telegram_test.go @@ -299,6 +299,81 @@ func TestSend_NonToolFeedbackDeletesTrackedProgressMessage(t *testing.T) { assert.False(t, ok, "tracked tool feedback should be cleared after final reply") } +func TestSend_ToolFeedbackTrackingIsTopicScoped(t *testing.T) { + nextMessageID := 0 + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + nextMessageID++ + return successResponseWithMessageID(t, nextMessageID), nil + }, + } + ch := newTestChannel(t, caller) + + _, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "-1001234567890", + Content: "🔧 `read_file`", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "-1001234567890", + TopicID: "42", + Raw: map[string]string{ + "message_kind": "tool_feedback", + }, + }, + }) + require.NoError(t, err) + + _, ok := ch.currentToolFeedbackMessage("-1001234567890") + assert.False(t, ok, "base chat should not track topic-specific tool feedback") + + msgID, ok := ch.currentToolFeedbackMessage("-1001234567890/42") + require.True(t, ok, "topic chat should track tool feedback") + assert.Equal(t, "1", msgID) +} + +func TestSend_TopicReplyDoesNotFinalizeDifferentTopicToolFeedback(t *testing.T) { + nextMessageID := 0 + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + nextMessageID++ + return successResponseWithMessageID(t, nextMessageID), nil + }, + } + ch := newTestChannel(t, caller) + + _, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "-1001234567890", + Content: "🔧 `read_file`", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "-1001234567890", + TopicID: "42", + Raw: map[string]string{ + "message_kind": "tool_feedback", + }, + }, + }) + require.NoError(t, err) + + ids, err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "-1001234567890", + Content: "final reply in another topic", + Context: bus.InboundContext{ + Channel: "telegram", + ChatID: "-1001234567890", + TopicID: "43", + }, + }) + require.NoError(t, err) + require.Len(t, caller.calls, 2) + assert.Equal(t, []string{"2"}, ids) + assert.Contains(t, caller.calls[1].URL, "sendMessage") + assert.NotContains(t, caller.calls[1].URL, "editMessageText") + + _, ok := ch.currentToolFeedbackMessage("-1001234567890/42") + assert.True(t, ok, "tool feedback in the original topic should remain tracked") +} + func TestFinalizeTrackedToolFeedbackMessage_StopsTrackingBeforeEdit(t *testing.T) { ch := newTestChannel(t, &stubCaller{ callFn: func(context.Context, string, *ta.RequestData) (*ta.Response, error) { @@ -660,6 +735,58 @@ func TestSend_UsesContextTopicIDWhenChatIDDoesNotIncludeThread(t *testing.T) { assert.Equal(t, "Hello from topic context", params.Text) } +func TestBeginStream_UpdateUsesForumThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return &ta.Response{Ok: true, Result: []byte("true")}, nil + }, + } + ch := newTestChannel(t, caller) + ch.tgCfg.Streaming.Enabled = true + + streamer, err := ch.BeginStream(context.Background(), "-1001234567890/42") + require.NoError(t, err) + require.NoError(t, streamer.Update(context.Background(), "partial")) + require.Len(t, caller.calls, 1) + assert.Contains(t, caller.calls[0].URL, "sendMessageDraft") + + var params struct { + ChatID int64 `json:"chat_id"` + MessageThreadID int `json:"message_thread_id"` + Text string `json:"text"` + } + require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, ¶ms)) + assert.Equal(t, int64(-1001234567890), params.ChatID) + assert.Equal(t, 42, params.MessageThreadID) + assert.Equal(t, "partial", params.Text) +} + +func TestBeginStream_FinalizeUsesForumThreadID(t *testing.T) { + caller := &stubCaller{ + callFn: func(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) { + return successResponse(t), nil + }, + } + ch := newTestChannel(t, caller) + ch.tgCfg.Streaming.Enabled = true + + streamer, err := ch.BeginStream(context.Background(), "-1001234567890/42") + require.NoError(t, err) + require.NoError(t, streamer.Finalize(context.Background(), "final")) + require.Len(t, caller.calls, 1) + assert.Contains(t, caller.calls[0].URL, "sendMessage") + + var params struct { + ChatID int64 `json:"chat_id"` + MessageThreadID int `json:"message_thread_id"` + Text string `json:"text"` + } + require.NoError(t, json.Unmarshal(caller.calls[0].Data.BodyRaw, ¶ms)) + assert.Equal(t, int64(-1001234567890), params.ChatID) + assert.Equal(t, 42, params.MessageThreadID) + assert.Equal(t, "final", params.Text) +} + func TestHandleMessage_ForumTopic_SetsMetadata(t *testing.T) { messageBus := bus.NewMessageBus() ch := &TelegramChannel{ diff --git a/pkg/providers/common/common.go b/pkg/providers/common/common.go index c167b1ffd..0a702e85e 100644 --- a/pkg/providers/common/common.go +++ b/pkg/providers/common/common.go @@ -240,8 +240,9 @@ func ParseResponse(body io.Reader) (*LLMResponse, error) { ID string `json:"id"` Type string `json:"type"` Function *struct { - Name string `json:"name"` - Arguments json.RawMessage `json:"arguments"` + Name string `json:"name"` + Arguments json.RawMessage `json:"arguments"` + ThoughtSignature string `json:"thought_signature"` } `json:"function"` ExtraContent *struct { Google *struct { @@ -273,9 +274,11 @@ func ParseResponse(body io.Reader) (*LLMResponse, error) { arguments := make(map[string]any) name := "" - // Extract thought_signature from Gemini/Google-specific extra content thoughtSignature := "" - if tc.ExtraContent != nil && tc.ExtraContent.Google != nil { + if tc.Function != nil { + thoughtSignature = tc.Function.ThoughtSignature + } + if thoughtSignature == "" && tc.ExtraContent != nil && tc.ExtraContent.Google != nil { thoughtSignature = tc.ExtraContent.Google.ThoughtSignature } @@ -291,9 +294,12 @@ func ParseResponse(body io.Reader) (*LLMResponse, error) { ThoughtSignature: thoughtSignature, } - if tc.ExtraContent != nil { + if thoughtSignature != "" || tc.ExtraContent != nil { extraContent := &ExtraContent{ - ToolFeedbackExplanation: tc.ExtraContent.ToolFeedbackExplanation, + ToolFeedbackExplanation: "", + } + if tc.ExtraContent != nil { + extraContent.ToolFeedbackExplanation = tc.ExtraContent.ToolFeedbackExplanation } if thoughtSignature != "" { extraContent.Google = &GoogleExtra{ diff --git a/pkg/providers/common/common_test.go b/pkg/providers/common/common_test.go index affb91e6f..a42d778f1 100644 --- a/pkg/providers/common/common_test.go +++ b/pkg/providers/common/common_test.go @@ -745,3 +745,27 @@ func TestParseResponse_WithThoughtSignature(t *testing.T) { out.ToolCalls[0].ExtraContent.Google.ThoughtSignature, "sig123") } } + +func TestParseResponse_WithFunctionThoughtSignature(t *testing.T) { + body := `{"choices":[{"message":{"content":"","tool_calls":[{"id":"call_1","type":"function","function":{"name":"test_tool","arguments":"{}","thought_signature":"sig456"}}]},"finish_reason":"tool_calls"}]}` + out, err := ParseResponse(strings.NewReader(body)) + if err != nil { + t.Fatalf("ParseResponse() error = %v", err) + } + if len(out.ToolCalls) != 1 { + t.Fatalf("len(ToolCalls) = %d, want 1", len(out.ToolCalls)) + } + if out.ToolCalls[0].ThoughtSignature != "sig456" { + t.Fatalf("ThoughtSignature = %q, want %q", out.ToolCalls[0].ThoughtSignature, "sig456") + } + if out.ToolCalls[0].ExtraContent == nil || out.ToolCalls[0].ExtraContent.Google == nil { + t.Fatal("ExtraContent.Google is nil") + } + if out.ToolCalls[0].ExtraContent.Google.ThoughtSignature != "sig456" { + t.Fatalf( + "ExtraContent.Google.ThoughtSignature = %q, want %q", + out.ToolCalls[0].ExtraContent.Google.ThoughtSignature, + "sig456", + ) + } +} diff --git a/web/backend/api/pico.go b/web/backend/api/pico.go index ffd0796c7..bb111b156 100644 --- a/web/backend/api/pico.go +++ b/web/backend/api/pico.go @@ -192,6 +192,10 @@ func (h *Handler) handleRegenPicoToken(w http.ResponseWriter, r *http.Request) { return } + gateway.mu.Lock() + gateway.picoToken = token + gateway.mu.Unlock() + h.writePicoInfoResponse(w, r, cfg, nil) } diff --git a/web/backend/api/pico_test.go b/web/backend/api/pico_test.go index a56cd9ba2..0efd5cd95 100644 --- a/web/backend/api/pico_test.go +++ b/web/backend/api/pico_test.go @@ -392,6 +392,58 @@ func TestHandleGetPicoInfo_OmitsToken(t *testing.T) { } } +func TestHandleRegenPicoToken_RefreshesGatewayTokenCache(t *testing.T) { + configPath := filepath.Join(t.TempDir(), "config.json") + h := NewHandler(configPath) + + if _, err := h.EnsurePicoChannel(); err != nil { + t.Fatalf("EnsurePicoChannel() error = %v", err) + } + + origPicoToken := gateway.picoToken + t.Cleanup(func() { + gateway.mu.Lock() + gateway.picoToken = origPicoToken + gateway.mu.Unlock() + }) + + gateway.mu.Lock() + gateway.picoToken = "stale-token" + gateway.mu.Unlock() + + req := httptest.NewRequest(http.MethodPost, "http://launcher.local/api/pico/token", nil) + rec := httptest.NewRecorder() + h.handleRegenPicoToken(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + + cfg, err := config.LoadConfig(configPath) + if err != nil { + t.Fatalf("LoadConfig() error = %v", err) + } + + bc := cfg.Channels["pico"] + decoded, err := bc.GetDecoded() + if err != nil { + t.Fatalf("GetDecoded() error = %v", err) + } + token := decoded.(*config.PicoSettings).Token.String() + if token == "" { + t.Fatal("expected regenerated pico token to be persisted") + } + if token == "stale-token" { + t.Fatal("expected regenerated pico token to differ from stale cache") + } + + gateway.mu.Lock() + defer gateway.mu.Unlock() + if gateway.picoToken != token { + t.Fatalf("gateway.picoToken = %q, want %q", gateway.picoToken, token) + } +} + func TestHandleWebSocketProxyReloadsGatewayTargetFromConfig(t *testing.T) { origMatcher := gatewayProcessMatcher gatewayProcessMatcher = func(int) (bool, bool) { return true, true }