diff --git a/CHANGELOG.md b/CHANGELOG.md index 367ca7fa8d..791fc97586 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased ### Fixed +- **Feishu recall fallback probes**: throttle repeated active-message recall checks so long-running turns do not continuously call platform message APIs. - **Skill discovery depth-1 only**: skill scanning no longer recurses into subdirectories. Only `//SKILL.md` is registered; nested SKILL.md files (e.g. inside `/references/...`) are treated as skill assets and ignored, matching the Claude Code CLI convention. Previously, nested SKILL.md files leaked into platform command menus as phantom slash commands (101 leaked commands from `frontend-design` skill alone) (#1304). - **feishu**: coalesce consecutive image messages from the same session into a single multi-image dispatch to fix first-image drop on batch sends (#1395). When the Feishu mobile client sends N images in quick succession, each image arrives as a separate `image` event with very close `create_time` values. Dispatching each immediately caused core/engine's `create_time` watermark (PR #1168) to drop the oldest image, so the agent only saw N-1 images. A per-session image buffer with a 150ms quiet window now merges the burst into one `core.Message` carrying all images, in send order. Single-image sends and quoted-image replies are unaffected. - **core**: queue post-restart notification and dispatch on platform ready (#1383). Previously `/restart` sent the success notification immediately after engine startup, racing the platform's async connect window (Telegram: ~2.6s). On a not-yet-ready platform the send was silently dropped at debug log level. The notify is now queued on the engine and dispatched when the target platform reaches `OnPlatformReady`, with bounded retry (3 attempts, 0/500/1500 ms backoff) on transient send failure. Failed sends log at warn level. A 10s safety timeout drops the notify with a warning if the target platform never reaches ready, so startup is never blocked indefinitely. Also covers Discord / Weixin / Matrix (other AsyncRecoverablePlatform implementations) for free. diff --git a/core/engine.go b/core/engine.go index 8b226709f0..db5d6de6d2 100644 --- a/core/engine.go +++ b/core/engine.go @@ -79,9 +79,10 @@ const ( ) const ( - messageRecallCheckTimeout = 2 * time.Second - messageRecallPollInterval = 2 * time.Second - recalledStopLockWait = 2 * time.Second + messageRecallCheckTimeout = 2 * time.Second + messageRecallPollInterval = 2 * time.Second + messageRecallProbeCooldown = time.Minute + recalledStopLockWait = 2 * time.Second ) // VersionInfo is set by main at startup so that /version works. @@ -494,25 +495,28 @@ type queuedMessage struct { // interactiveState tracks a running interactive agent session and its permission state. type interactiveState struct { - agentSession AgentSession - platform Platform - replyCtx any - currentMessageID string - workspaceDir string - agent Agent - mu sync.Mutex - stopCh chan struct{} - stopped bool - pending *pendingPermission - pendingMessages []queuedMessage // messages queued while session was busy - approveAll bool // when true, auto-approve all permission requests for this session - fromVoice bool // true if current turn originated from voice transcription - sideText string - deleteMode *deleteModeState - modelSwitch *modelSwitchState - pendingProviderAdd *pendingProviderAddState - lastAutoCompressAt time.Time - lastAutoCompressTokens int + agentSession AgentSession + platform Platform + replyCtx any + currentMessageID string + lastRecallProbeMessageID string + lastRecallProbeAt time.Time + recallProbeInFlight bool + workspaceDir string + agent Agent + mu sync.Mutex + stopCh chan struct{} + stopped bool + pending *pendingPermission + pendingMessages []queuedMessage // messages queued while session was busy + approveAll bool // when true, auto-approve all permission requests for this session + fromVoice bool // true if current turn originated from voice transcription + sideText string + deleteMode *deleteModeState + modelSwitch *modelSwitchState + pendingProviderAdd *pendingProviderAddState + lastAutoCompressAt time.Time + lastAutoCompressTokens int // Unsolicited event reader: a background goroutine that consumes agent // events between user-initiated turns (e.g. background task completions). @@ -2594,15 +2598,48 @@ func (e *Engine) stopCurrentMessageIfRecalled(sessionKey string) bool { platform := state.platform replyCtx := state.replyCtx messageID := state.currentMessageID - state.mu.Unlock() if platform == nil || replyCtx == nil || messageID == "" { + state.mu.Unlock() return false } - detector, ok := platform.(MessageRecallDetector) if !ok { + state.mu.Unlock() + return false + } + if state.recallProbeInFlight { + state.mu.Unlock() + slog.Debug("message recall fallback probe skipped; probe already in flight", + "platform", platform.Name(), + "msg_id", messageID, + "session", sessionKey, + ) return false } + now := time.Now() + if state.lastRecallProbeMessageID == messageID && now.Sub(state.lastRecallProbeAt) < messageRecallProbeCooldown { + nextProbeIn := messageRecallProbeCooldown - now.Sub(state.lastRecallProbeAt) + state.mu.Unlock() + slog.Debug("message recall fallback probe throttled", + "platform", platform.Name(), + "msg_id", messageID, + "session", sessionKey, + "next_probe_in", nextProbeIn, + ) + return false + } + state.recallProbeInFlight = true + state.lastRecallProbeMessageID = messageID + state.lastRecallProbeAt = now + state.mu.Unlock() + + defer func() { + state.mu.Lock() + if state.currentMessageID == messageID { + state.recallProbeInFlight = false + } + state.mu.Unlock() + }() ctx, cancel := context.WithTimeout(e.ctx, messageRecallCheckTimeout) defer cancel() @@ -3587,6 +3624,11 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session // Update reply context for this turn state.mu.Lock() + if state.currentMessageID != msg.MessageID { + state.lastRecallProbeMessageID = "" + state.lastRecallProbeAt = time.Time{} + state.recallProbeInFlight = false + } state.platform = p state.replyCtx = msg.ReplyCtx state.currentMessageID = msg.MessageID diff --git a/core/engine_test.go b/core/engine_test.go index 0fc73ee4b5..7cd2585606 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -10060,6 +10060,50 @@ func TestHandleMessageBusyRecalledCurrentStopsAndProcessesNewMessage(t *testing. } } +func TestStopCurrentMessageIfRecalledThrottlesRepeatedFallbackChecks(t *testing.T) { + p := &recallCheckingPlatform{ + stubPlatformEngine: stubPlatformEngine{n: "test"}, + recalled: false, + } + e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish) + key := "test:user1" + state := &interactiveState{ + agentSession: newControllableSession("current"), + platform: p, + replyCtx: "reply-ctx-1", + currentMessageID: "msg-1", + } + e.interactiveMu.Lock() + e.interactiveStates[key] = state + e.interactiveMu.Unlock() + + for range 3 { + if e.stopCurrentMessageIfRecalled(key) { + t.Fatal("stopCurrentMessageIfRecalled returned true for non-recalled message") + } + } + checked := p.checkedReplyCtxs() + if len(checked) != 1 || checked[0] != "reply-ctx-1" { + t.Fatalf("checked reply contexts = %v, want exactly one check for reply-ctx-1", checked) + } + + state.mu.Lock() + state.replyCtx = "reply-ctx-2" + state.currentMessageID = "msg-2" + state.lastRecallProbeMessageID = "" + state.lastRecallProbeAt = time.Time{} + state.recallProbeInFlight = false + state.mu.Unlock() + + if e.stopCurrentMessageIfRecalled(key) { + t.Fatal("stopCurrentMessageIfRecalled returned true for second non-recalled message") + } + checked = p.checkedReplyCtxs() + if len(checked) != 2 || checked[1] != "reply-ctx-2" { + t.Fatalf("checked reply contexts = %v, want second check for new message", checked) + } +} + func TestExecuteCardAction_NewCleansUpAndCreatesSession(t *testing.T) { p := &stubPlatformEngine{n: "test"} e := NewEngine("test", &stubAgent{}, []Platform{p}, "", LangEnglish)