diff --git a/README.md b/README.md index 14a03e5..1eab6c3 100644 --- a/README.md +++ b/README.md @@ -76,8 +76,82 @@ Each bot is one JSON file in `config/` (see `config/default.json` for the templa | `max_tokens` | number | `1000` | Maximum output tokens per reply. **Thinking tokens count toward this limit** — raise it (e.g. `4000`+) whenever `thinking` is `"adaptive"`, or a turn can spend the whole budget on reasoning and produce no text. | | `thinking` | string | *(omitted)* | Reasoning mode: `"adaptive"` (the model decides when and how much to think) or `"disabled"`. Omit the key entirely to use the model's own API default. If the configured model doesn't support the chosen mode, the API rejects the request with a 400 — owners/admins see the raw error, regular users get the generic fallback. Check [Anthropic's model docs](https://platform.claude.com/docs/en/build-with-claude/adaptive-thinking) for per-model support. | | `thinking_display` | string | *(omitted)* | `"summarized"` or `"omitted"`. Only valid together with `"thinking": "adaptive"`. Controls whether the API returns a readable summary of the reasoning (logged, never sent to chat). Thinking is billed the same either way; when omitted, the API's per-model default applies. | +| `debounce_ms` | number | *(omitted)* | Quiet window, in milliseconds, for coalescing rapid follow-up messages into a single turn. Omit or set `0` to disable. See [Coalescing rapid messages](#coalescing-rapid-messages) below. | +| `cache_history` | bool | `true` | Places a prompt-cache breakpoint on the trailing conversation block, so each turn reads the prior history from cache instead of reprocessing it at full price. Set `false` to cache only the system prompt. | +| `web_search` | object | *(omitted)* | Enables Anthropic's server-side web search (and, optionally, web fetch), sandboxed to a domain allowlist. Omit the key entirely to leave both tools off — an absent block sends byte-identical requests to before. See [Web search / fetch](#web-search--fetch) below. | -Every reply logs one accounting line — `[usage] model=... in=... out=... thinking=... stop=...` — so thinking spend (billed even when its text is omitted) stays visible in `journalctl`/`docker compose logs`. A `stop=max_tokens` line is accompanied by an error-level warning that the reply was truncated. +Every reply logs one accounting line — `[usage] model=... in=... out=... thinking=... cache_read=... cache_write=... stop=...` — so thinking spend (billed even when its text is omitted) stays visible in `journalctl`/`docker compose logs`. A `stop=max_tokens` line is accompanied by an error-level warning that the reply was truncated. + +> [!TIP] +> Watch `cache_read`/`cache_write` after changing prompts or models. A cache breakpoint below the model's minimum cacheable prefix fails **silently** — no error, just `cache_write=0` forever. The minimum is model-specific and not monotonic across generations (Haiku 4.5 needs 4096 tokens; Sonnet 4.6 needs 1024), so a short system prompt that caches fine on one model may never cache on another. Note also that chat memory is a sliding window: once it is full, each turn evicts the oldest message and changes the prefix, so `cache_read` on long-running chats will be lower than on fresh ones. + +### Coalescing rapid messages + +A user who sends "how do I do X", then "sorry typo", then "lmao" in five seconds would otherwise get three separate replies — the bot starts a full turn per message, because each Telegram update independently drives one. `debounce_ms` holds text messages in a per-chat buffer and resets the window on every new message, dispatching a single turn once the user stops typing. + +```json +"debounce_ms": 2500 +``` + +Reasonable windows are 1500–3000ms for ordinary chat and up to 8000ms for Telegram Business, where a person writing to a business account tends to send longer bursts. The maximum is 30000ms. + +Nothing is discarded while buffering. Each message is still persisted and added to chat memory the moment it arrives, so the single coalesced turn sees all of them — the buffer only decides *when* to answer, never *what* the model reads. Reply metadata (language, premium status, business connection) follows the most recent message in the batch. + +What deliberately does **not** wait: + +- **Commands** (`/stats`, `/clear`, …) dispatch immediately. +- **Photos, albums, voice, and stickers** dispatch immediately and cancel any pending text window. The buffered text is not lost — it is already in memory, so the media turn answers it too. +- **`/clear` and `/clear_hard`** cancel the buffer outright. Without this, the window would fire seconds after the wipe and replay the just-deleted messages back into memory. + +While a turn is running the bot shows Telegram's "typing…" indicator (or "recording audio" while synthesising a voice reply), refreshed every 4 seconds because Telegram expires the status after 5. Without it, a debounce window reads as the bot ignoring you — which is what prompts users to send more messages in the first place. + +> [!NOTE] +> Debouncing is the cheap fix and the reason there is no "cancel the in-flight request" mode. Anthropic bills input tokens plus any output already generated when a turn stops partway, and any web searches it already ran are billed and re-billed on the retry. A message that never dispatched costs nothing. + +### Web search / fetch + +The `web_search` block wires in Anthropic's server-side `web_search` (and optionally `web_fetch`) tools, mirroring how `mcp_servers` wires in MCP toolsets: the engine provides the generic mechanism, and the per-bot config carries the policy. **No domains are hardcoded in the engine** — each profile declares its own allowlist. When the model decides a question needs current information, it runs a search server-side; results and any fetch targets are confined to the domains you list. + +```json +"web_search": { + "allowed_domains": [ + "example.com/help", + "docs.example.com" + ], + "max_uses": 3, + "fetch": true, + "max_content_tokens": 50000 +} +``` + +| Field | Type | Description | +| ---------------------- | -------- | ----------- | +| `allowed_domains` | string[] | Restricts **`web_search`** results (and, by default, `web_fetch` targets) to these. Each entry is a **domain with an optional path prefix and no scheme** — `example.com` or `example.com/help`. Mutually exclusive with `blocked_domains`. **The path only scopes `web_search`** — see the asymmetry note below. | +| `blocked_domains` | string[] | Excludes these domains instead of allowlisting. Mutually exclusive with `allowed_domains` (setting both fails validation at boot, since the API 400s). | +| `fetch_allowed_domains`| string[] | **Host-only** allowlist for `web_fetch`, independent of the search list. Omit it and `web_fetch` reuses the *hosts* of `allowed_domains`. Set it to fetch a **narrower** set than you search — e.g. to keep a shared host (a social platform) search-only. Only meaningful with `fetch: true`. | +| `max_uses` | number | Caps how many searches the model may run per turn. Omit for no cap. | +| `fetch` | bool | Also enable `web_fetch`, which pulls full page content for URLs already surfaced by a search or pasted by the user (it cannot fetch model-invented URLs). Citations are always on when fetch is enabled, so fetched passages are sourceable. | +| `max_content_tokens` | number | Caps the tokens a single `web_fetch` may pull into context. Only meaningful with `fetch: true`. | + +**Search/fetch path asymmetry (important).** Anthropic filters the two tools differently: `web_search` honors a path prefix (`example.com/help` matches only `example.com/help/...`), but **`web_fetch` matches on the host only — an entry that includes a path never matches any fetch URL**. The engine bridges this: `web_search` gets your entries verbatim, while `web_fetch` gets the host portion of each entry (path stripped, deduped). So a path-scoped `allowed_domains` still permits fetching across the whole host. If that host isn't wholly trusted, list the fetchable hosts explicitly in `fetch_allowed_domains` instead. + +**Search-only pattern for shared hosts (e.g. social).** To let the model *search* a single account on a shared platform without ever *fetching* the wider platform, put the account path in `allowed_domains` and leave its host out of `fetch_allowed_domains`: + +```json +"web_search": { + "allowed_domains": ["helpcenter.example/hc", "x.com/youraccount"], + "fetch_allowed_domains": ["helpcenter.example"], + "fetch": true +} +``` + +Here search is confined to your help center *and* your one social account, but fetch can only ever reach `helpcenter.example` — a pasted `x.com/someone-else` URL is not fetchable. Note the search side only sandboxes cleanly on platforms that keep an account's content under its handle path (X/Twitter, TikTok `@handle`, Facebook); use the `handle/` or `handle/*` form to avoid a prefix bleed. Instagram is an exception — posts live at `instagram.com/p/`, not under the handle — so it can't be account-sandboxed by path. + +The allowlist is a **hard, server-side sandbox**, not a prompt request — off-list results are dropped and an off-list fetch target returns `url_not_allowed`, which is the exfiltration mitigation Anthropic recommends for bots processing untrusted input. Keep the list tight. A fetch can still fail for site-side reasons (bot protection such as Cloudflare returns `url_not_accessible`); the search snippet plus citations remain the reliable signal, so a full fetch is a bonus, not a dependency. + +`web_fetch` can only open a URL that a `web_search` result surfaced (or the user pasted) — never one the model invents or rebuilds from a page title; those return `url_not_in_prior_context`. So a fetch-enabled bot's prompt should steer it to **search first, then fetch a returned URL**, and to **re-search with a more specific query** (rather than guess a URL) when the exact page it wants isn't in the results. + +Web search is **not free** — Anthropic bills per search (plus the tokens the results add) — so scope the allowlist and `max_uses` deliberately, and lean on a prompt that tells the bot to search only when a question actually turns on current or authoritative information. Search activity logs as `[web] ...` lines, and long multi-search turns are handled transparently (the engine follows Anthropic's `pause_turn` continuations up to a small cap, so a slow search doesn't truncate the reply). > [!TIP] > For deep request/response debugging, the Anthropic Go SDK ships `option.WithDebugLog(...)` (dumps full HTTP bodies with auth headers redacted). It is not wired into the bot — dev-only, add it temporarily to the client constructor if you ever need wire-level traces. diff --git a/anthropic.go b/anthropic.go index f6fa8da..02d8cd2 100644 --- a/anthropic.go +++ b/anthropic.go @@ -17,6 +17,8 @@ var ErrModelNotFound = errors.New("model not found or deprecated") const maxFileNotFoundRetries = 3 +const maxPauseTurnContinuations = 5 + const defaultMaxTokens = 1000 const mcpUnsupportedSentinel = "format not currently supported by the Anthropic API" @@ -65,9 +67,10 @@ func (b *Bot) getAnthropicResponse(ctx context.Context, chatID int64, messages [ params.Thinking = thinking } + var tools []anthropic.BetaToolUnionParam + if len(b.config.MCPServers) > 0 { mcpServers := make([]anthropic.BetaRequestMCPServerURLDefinitionParam, 0, len(b.config.MCPServers)) - tools := make([]anthropic.BetaToolUnionParam, 0, len(b.config.MCPServers)) for _, s := range b.config.MCPServers { srv := anthropic.BetaRequestMCPServerURLDefinitionParam{ Name: s.Name, @@ -95,32 +98,126 @@ func (b *Bot) getAnthropicResponse(ctx context.Context, chatID int64, messages [ tools = append(tools, anthropic.BetaToolUnionParam{OfMCPToolset: toolset}) } params.MCPServers = mcpServers - params.Tools = tools params.Betas = append(params.Betas, anthropic.AnthropicBetaMCPClient2025_11_20) } - for attempt := 0; attempt < maxFileNotFoundRetries; attempt++ { - joined, streamErr := b.streamMessages(ctx, params, onSegment) - if streamErr == nil { - return joined, nil - } - var apiErr *anthropic.Error - if !errors.As(streamErr, &apiErr) || apiErr.StatusCode != http.StatusNotFound { - return "", fmt.Errorf("error creating Anthropic message: %w", streamErr) - } - missingFileID := extractMissingFileID(streamErr) - if missingFileID == "" { - return "", fmt.Errorf("%w: %s", ErrModelNotFound, b.config.Model) - } - ErrorLogger.Printf("[%s] self-heal: stripping dead file_id %s from chat %d (attempt %d/%d)", - b.config.ID, missingFileID, chatID, attempt+1, maxFileNotFoundRetries) - b.stripDeadFileIDFromMemory(chatID, missingFileID) - if _, cleanupErr := b.markFilesPendingCleanup(ctx, chatID, []string{missingFileID}); cleanupErr != nil { - ErrorLogger.Printf("[%s] mark files pending cleanup: %v", b.config.ID, cleanupErr) - } - params.Messages = b.prepareContextMessages(b.getOrCreateChatMemory(chatID)) + tools = append(tools, webSearchTools(b.config.WebSearch)...) + + if len(tools) > 0 { + params.Tools = tools } - return "", fmt.Errorf("max self-heal retries (%d) exceeded: too many file_ids gone from anthropic", maxFileNotFoundRetries) + + var fullText strings.Builder + var lastMsg anthropic.BetaMessage + fileRetries, pauseContinuations := 0, 0 + for { + joined, msg, streamErr := b.streamMessages(ctx, params, onSegment) + if streamErr != nil { + var apiErr *anthropic.Error + if !errors.As(streamErr, &apiErr) || apiErr.StatusCode != http.StatusNotFound { + return "", fmt.Errorf("error creating Anthropic message: %w", streamErr) + } + missingFileID := extractMissingFileID(streamErr) + if missingFileID == "" { + return "", fmt.Errorf("%w: %s", ErrModelNotFound, b.config.Model) + } + fileRetries++ + if fileRetries > maxFileNotFoundRetries { + return "", fmt.Errorf("max self-heal retries (%d) exceeded: too many file_ids gone from anthropic", maxFileNotFoundRetries) + } + ErrorLogger.Printf("[%s] self-heal: stripping dead file_id %s from chat %d (attempt %d/%d)", + b.config.ID, missingFileID, chatID, fileRetries, maxFileNotFoundRetries) + b.stripDeadFileIDFromMemory(chatID, missingFileID) + if _, cleanupErr := b.markFilesPendingCleanup(ctx, chatID, []string{missingFileID}); cleanupErr != nil { + ErrorLogger.Printf("[%s] mark files pending cleanup: %v", b.config.ID, cleanupErr) + } + params.Messages = b.prepareContextMessages(b.getOrCreateChatMemory(chatID)) + continue + } + + lastMsg = msg + if joined != "" { + if fullText.Len() > 0 { + fullText.WriteString("\n\n") + } + fullText.WriteString(joined) + } + + if msg.StopReason == anthropic.BetaStopReasonPauseTurn { + pauseContinuations++ + if pauseContinuations > maxPauseTurnContinuations { + ErrorLogger.Printf("[%s] pause_turn continuations exceeded (%d); returning partial answer", + b.config.ID, maxPauseTurnContinuations) + break + } + params.Messages = append(params.Messages, msg.ToParam()) + continue + } + break + } + + if fullText.Len() == 0 { + return "", emptyStreamError(string(lastMsg.StopReason), + lastMsg.Usage.OutputTokensDetails.ThinkingTokens, params.MaxTokens) + } + return fullText.String(), nil +} + +func webSearchTools(cfg *WebSearchConfig) []anthropic.BetaToolUnionParam { + if cfg == nil { + return nil + } + search := &anthropic.BetaWebSearchTool20250305Param{ + AllowedDomains: cfg.AllowedDomains, + BlockedDomains: cfg.BlockedDomains, + } + if cfg.MaxUses > 0 { + search.MaxUses = param.NewOpt(int64(cfg.MaxUses)) + } + tools := []anthropic.BetaToolUnionParam{{OfWebSearchTool20250305: search}} + + if cfg.Fetch { + fetchAllowed := cfg.FetchAllowedDomains + if len(fetchAllowed) == 0 { + fetchAllowed = cfg.AllowedDomains + } + fetch := &anthropic.BetaWebFetchTool20250910Param{ + AllowedDomains: fetchHosts(fetchAllowed), + BlockedDomains: fetchHosts(cfg.BlockedDomains), + Citations: anthropic.BetaCitationsConfigParam{Enabled: param.NewOpt(true)}, + } + if cfg.MaxUses > 0 { + fetch.MaxUses = param.NewOpt(int64(cfg.MaxUses)) + } + if cfg.MaxContentTokens > 0 { + fetch.MaxContentTokens = param.NewOpt(int64(cfg.MaxContentTokens)) + } + tools = append(tools, anthropic.BetaToolUnionParam{OfWebFetchTool20250910: fetch}) + } + return tools +} + +// fetchHosts strips any path from each domain entry. web_fetch matches on host +// only, so a path-scoped entry (valid for web_search) would otherwise never match +// any fetch URL. Search keeps the path-scoped entries; fetch gets host-only. +func fetchHosts(entries []string) []string { + if len(entries) == 0 { + return nil + } + seen := make(map[string]bool, len(entries)) + hosts := make([]string, 0, len(entries)) + for _, e := range entries { + host := e + if i := strings.IndexByte(host, '/'); i >= 0 { + host = host[:i] + } + if host == "" || seen[host] { + continue + } + seen[host] = true + hosts = append(hosts, host) + } + return hosts } func buildUserContext(username, firstName, lastName string, isPremium bool, languageCode string, messageTime int) string { @@ -179,7 +276,7 @@ func thinkingParamFromConfig(mode, display string) (anthropic.BetaThinkingConfig } } -func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNewParams, onSegment func(string) error) (string, error) { +func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNewParams, onSegment func(string) error) (string, anthropic.BetaMessage, error) { stream := b.anthropicClient.Beta.Messages.NewStreaming(ctx, params) defer func() { if err := stream.Close(); err != nil { @@ -188,6 +285,7 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe }() var ( + message anthropic.BetaMessage allSegments []string currentKind string currentText strings.Builder @@ -197,31 +295,24 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe currentTResultUseID, currentTResultServer string currentTResultIsError bool currentTResultContent string + currentServerToolName, currentServerToolID string + currentServerResult string mcpCalls = map[string]mcpCall{} - startInputTokens int64 - finalUsage anthropic.BetaMessageDeltaUsage - stopReason string ) for stream.Next() { e := stream.Current() + if accErr := message.Accumulate(e); accErr != nil { + ErrorLogger.Printf("[stream] accumulate failed: %v", accErr) + } switch e.Type { - case "message_start": - startInputTokens = e.AsMessageStart().Message.Usage.InputTokens - - case "message_delta": - md := e.AsMessageDelta() - finalUsage = md.Usage - if md.Delta.StopReason != "" { - stopReason = string(md.Delta.StopReason) - } - case "content_block_start": cbs := e.AsContentBlockStart() currentKind = cbs.ContentBlock.Type currentText.Reset() currentThinking.Reset() currentInputJSON.Reset() + currentServerResult = "" switch currentKind { case "mcp_tool_use": currentTUseName = cbs.ContentBlock.Name @@ -232,6 +323,11 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe currentTResultServer = cbs.ContentBlock.ServerName currentTResultIsError = cbs.ContentBlock.IsError currentTResultContent = cbs.ContentBlock.JSON.Content.Raw() + case "server_tool_use": + currentServerToolName = cbs.ContentBlock.Name + currentServerToolID = cbs.ContentBlock.ID + case "web_search_tool_result", "web_fetch_tool_result": + currentServerResult = cbs.ContentBlock.JSON.Content.Raw() } case "content_block_delta": @@ -246,7 +342,7 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe currentThinking.WriteString(cbd.Delta.Thinking) } case "input_json_delta": - if currentKind == "mcp_tool_use" { + if currentKind == "mcp_tool_use" || currentKind == "server_tool_use" { currentInputJSON.WriteString(cbd.Delta.PartialJSON) } } @@ -285,6 +381,15 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe "(total=%d): server=%q tool=%q input=%s tool_use_id=%q", b.config.ID, total, call.server, call.name, call.input, currentTResultUseID) } + case "server_tool_use": + InfoLogger.Printf("[web] %s id=%q input=%s", + currentServerToolName, currentServerToolID, currentInputJSON.String()) + case "web_search_tool_result", "web_fetch_tool_result": + preview := currentServerResult + if len(preview) > 500 { + preview = preview[:500] + "...(truncated)" + } + InfoLogger.Printf("[web] %s content=%s", currentKind, preview) case "thinking", "redacted_thinking": if summary := strings.TrimSpace(currentThinking.String()); summary != "" { if len(summary) > 500 { @@ -296,7 +401,7 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe } default: if currentKind != "" { - InfoLogger.Printf("[mcp] block type=%q (unhandled)", currentKind) + InfoLogger.Printf("[stream] block type=%q (unhandled)", currentKind) } } currentKind = "" @@ -304,27 +409,26 @@ func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNe } if err := stream.Err(); err != nil { - return "", err + return "", message, err } - if stopReason != "" || finalUsage.OutputTokens > 0 { - inputTokens := finalUsage.InputTokens - if inputTokens == 0 { - inputTokens = startInputTokens - } - InfoLogger.Printf("[usage] model=%s in=%d out=%d thinking=%d stop=%s", - params.Model, inputTokens, finalUsage.OutputTokens, - finalUsage.OutputTokensDetails.ThinkingTokens, stopReason) - if stopReason == "max_tokens" { + stopReason := string(message.StopReason) + if stopReason != "" || message.Usage.OutputTokens > 0 { + // cache_read/cache_write make the caching configuration falsifiable: a + // breakpoint below the model's minimum cacheable prefix fails silently, + // reporting cache_write=0 rather than raising an error. + InfoLogger.Printf("[usage] model=%s in=%d out=%d thinking=%d cache_read=%d cache_write=%d stop=%s", + params.Model, message.Usage.InputTokens, message.Usage.OutputTokens, + message.Usage.OutputTokensDetails.ThinkingTokens, + message.Usage.CacheReadInputTokens, message.Usage.CacheCreationInputTokens, + stopReason) + if message.StopReason == anthropic.BetaStopReasonMaxTokens { ErrorLogger.Printf("[usage] response truncated at max_tokens=%d - raise max_tokens (thinking counts toward it)", params.MaxTokens) } } - if len(allSegments) == 0 { - return "", emptyStreamError(stopReason, finalUsage.OutputTokensDetails.ThinkingTokens, params.MaxTokens) - } - return strings.Join(allSegments, "\n\n"), nil + return strings.Join(allSegments, "\n\n"), message, nil } func emptyStreamError(stopReason string, thinkingTokens, maxTokens int64) error { diff --git a/anthropic_test.go b/anthropic_test.go index 575fcae..b3fe3cf 100644 --- a/anthropic_test.go +++ b/anthropic_test.go @@ -127,6 +127,173 @@ func TestBackwardCompatibleParams(t *testing.T) { } } +func TestWebSearchTools(t *testing.T) { + t.Run("nil config yields no tools", func(t *testing.T) { + if tools := webSearchTools(nil); tools != nil { + t.Errorf("webSearchTools(nil) = %v, want nil", tools) + } + }) + + t.Run("search only when fetch off", func(t *testing.T) { + tools := webSearchTools(&WebSearchConfig{ + AllowedDomains: []string{"example.com/hc"}, + MaxUses: 3, + }) + if len(tools) != 1 { + t.Fatalf("got %d tools, want 1 (search only)", len(tools)) + } + if tools[0].OfWebSearchTool20250305 == nil { + t.Fatalf("tools[0] is not a web_search tool") + } + if tools[0].OfWebFetchTool20250910 != nil { + t.Error("web_fetch tool present but fetch is off") + } + }) + + t.Run("search + fetch with allowlist and citations", func(t *testing.T) { + tools := webSearchTools(&WebSearchConfig{ + AllowedDomains: []string{"example.com/hc", "docs.example.com"}, + MaxUses: 3, + Fetch: true, + MaxContentTokens: 50000, + }) + if len(tools) != 2 { + t.Fatalf("got %d tools, want 2 (search + fetch)", len(tools)) + } + + search := tools[0].OfWebSearchTool20250305 + if search == nil { + t.Fatalf("tools[0] is not a web_search tool") + } + if !sameStrings(search.AllowedDomains, []string{"example.com/hc", "docs.example.com"}) { + t.Errorf("search AllowedDomains = %v, want the path-scoped list unchanged", search.AllowedDomains) + } + if search.MaxUses.Value != 3 { + t.Errorf("search MaxUses = %d, want 3", search.MaxUses.Value) + } + + fetch := tools[1].OfWebFetchTool20250910 + if fetch == nil { + t.Fatalf("tools[1] is not a web_fetch tool") + } + if !sameStrings(fetch.AllowedDomains, []string{"example.com", "docs.example.com"}) { + t.Errorf("fetch AllowedDomains = %v, want host-only [example.com docs.example.com]", fetch.AllowedDomains) + } + if fetch.MaxContentTokens.Value != 50000 { + t.Errorf("fetch MaxContentTokens = %d, want 50000", fetch.MaxContentTokens.Value) + } + if fetch.Citations.Enabled.Value != true { + t.Error("fetch citations not enabled") + } + }) + + t.Run("fetch hosts are deduped", func(t *testing.T) { + tools := webSearchTools(&WebSearchConfig{ + AllowedDomains: []string{"a.com/x", "a.com/y", "b.com"}, + Fetch: true, + }) + fetch := tools[1].OfWebFetchTool20250910 + if fetch == nil { + t.Fatalf("tools[1] is not a web_fetch tool") + } + if !sameStrings(fetch.AllowedDomains, []string{"a.com", "b.com"}) { + t.Errorf("fetch AllowedDomains = %v, want deduped [a.com b.com]", fetch.AllowedDomains) + } + }) + + t.Run("social host is search-only via fetch_allowed_domains", func(t *testing.T) { + tools := webSearchTools(&WebSearchConfig{ + AllowedDomains: []string{"helpshift.example/hc", "x.com/thatskygame"}, + FetchAllowedDomains: []string{"helpshift.example"}, + Fetch: true, + }) + search := tools[0].OfWebSearchTool20250305 + if search == nil { + t.Fatalf("tools[0] is not a web_search tool") + } + var searchHasSocial bool + for _, d := range search.AllowedDomains { + if d == "x.com/thatskygame" { + searchHasSocial = true + } + } + if !searchHasSocial { + t.Errorf("search AllowedDomains = %v, want it to include x.com/thatskygame", search.AllowedDomains) + } + + fetch := tools[1].OfWebFetchTool20250910 + if fetch == nil { + t.Fatalf("tools[1] is not a web_fetch tool") + } + if !sameStrings(fetch.AllowedDomains, []string{"helpshift.example"}) { + t.Errorf("fetch AllowedDomains = %v, want only [helpshift.example]", fetch.AllowedDomains) + } + for _, d := range fetch.AllowedDomains { + if strings.HasPrefix(d, "x.com") { + t.Errorf("fetch AllowedDomains leaked the social host: %v", fetch.AllowedDomains) + } + } + }) + + t.Run("wire shape carries allowed_domains", func(t *testing.T) { + tools := webSearchTools(&WebSearchConfig{ + AllowedDomains: []string{"thatgamecompany.helpshift.com/hc"}, + MaxUses: 2, + Fetch: true, + }) + raw, err := json.Marshal(tools) + if err != nil { + t.Fatalf("marshal: %v", err) + } + body := string(raw) + for _, want := range []string{ + "web_search_20250305", + "web_fetch_20250910", + "thatgamecompany.helpshift.com/hc", + "allowed_domains", + } { + if !strings.Contains(body, want) { + t.Errorf("wire body missing %q:\n%s", want, body) + } + } + }) +} + +func sameStrings(got, want []string) bool { + if len(got) != len(want) { + return false + } + for i := range got { + if got[i] != want[i] { + return false + } + } + return true +} + +func TestFetchHosts(t *testing.T) { + cases := []struct { + name string + in []string + want []string + }{ + {"nil in nil out", nil, nil}, + {"empty in nil out", []string{}, nil}, + {"host passthrough", []string{"example.com"}, []string{"example.com"}}, + {"strip path", []string{"example.com/hc/en"}, []string{"example.com"}}, + {"dedup after strip", []string{"a.com/x", "a.com/y"}, []string{"a.com"}}, + {"preserve order and subdomains", []string{"docs.example.com/a", "example.com"}, []string{"docs.example.com", "example.com"}}, + {"drop empty leading slash", []string{"/oops", "ok.com"}, []string{"ok.com"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := fetchHosts(tc.in); !sameStrings(got, tc.want) { + t.Errorf("fetchHosts(%v) = %v, want %v", tc.in, got, tc.want) + } + }) + } +} + func TestEmptyStreamError(t *testing.T) { err := emptyStreamError("max_tokens", 3900, 4000) for _, want := range []string{"output budget exhausted", "3900", "4000"} { diff --git a/bot.go b/bot.go index 5125d6d..4a155d2 100644 --- a/bot.go +++ b/bot.go @@ -29,6 +29,9 @@ type Bot struct { botID uint albumBuffers map[string]*pendingAlbum albumBuffersMu sync.Mutex + intakeBuffers map[int64]*pendingIntake + intakeBuffersMu sync.Mutex + intakeSeq uint64 } func messageType(msg *models.Message) string { @@ -90,6 +93,7 @@ func NewBot(db *gorm.DB, config BotConfig, clock Clock, tgClient TelegramClient) botID: botEntry.ID, tgBot: tgClient, albumBuffers: make(map[string]*pendingAlbum), + intakeBuffers: make(map[int64]*pendingIntake), } if tgClient == nil { @@ -298,9 +302,39 @@ func (b *Bot) prepareContextMessages(chatMemory *ChatMemory) []anthropic.BetaMes } contextMessages = append(contextMessages, param) } + + if b.config.CacheHistoryEnabled() { + markTrailingCacheBreakpoint(contextMessages) + } return contextMessages } +// markTrailingCacheBreakpoint puts a cache_control breakpoint on the final +// content block of the conversation, so the next turn reads the whole prefix +// from cache instead of reprocessing it. The system prompt keeps its own +// breakpoint; tools and system render ahead of messages, so the two compose. +// +// Caveat worth knowing when reading [usage] lines: chat memory is a sliding +// window. Once it is full, each new turn evicts the oldest message, which +// changes the prefix and forces a miss. Until then, and for any chat shorter +// than the window, this converts a full-price reread into a cache read. +func markTrailingCacheBreakpoint(messages []anthropic.BetaMessageParam) { + if len(messages) == 0 { + return + } + blocks := messages[len(messages)-1].Content + if len(blocks) == 0 { + return + } + + switch last := &blocks[len(blocks)-1]; { + case last.OfText != nil: + last.OfText.CacheControl = anthropic.NewBetaCacheControlEphemeralParam() + case last.OfImage != nil: + last.OfImage.CacheControl = anthropic.NewBetaCacheControlEphemeralParam() + } +} + func contentBlocksForMessage(msg Message) []anthropic.BetaContentBlockParamUnion { var blocks []anthropic.BetaContentBlockParamUnion if msg.IsUser && len(msg.ImageFileIDs) > 0 { @@ -706,16 +740,18 @@ func (b *Bot) screenOutgoingMessage(chatID int64, response string) (Message, err return Message{}, err } + // Mark every outstanding user message in the chat, not just the newest one. + // A coalesced turn answers the whole batch, so a single-row update would + // leave the earlier messages permanently unanswered. This also drops an + // UPDATE ... ORDER BY ... LIMIT, which stock SQLite builds do not support. now := time.Now() err := b.db.Model(&Message{}). Where("chat_id = ? AND bot_id = ? AND is_user = ? AND answered_on IS NULL", chatID, b.botID, true). - Order("timestamp DESC"). - Limit(1). Update("answered_on", now).Error if err != nil { - ErrorLogger.Printf("Error marking user message as answered: %v", err) + ErrorLogger.Printf("Error marking user messages as answered: %v", err) } chatMemory := b.getOrCreateChatMemory(chatID) diff --git a/config.go b/config.go index 7e04315..587e21c 100644 --- a/config.go +++ b/config.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strings" + "time" ) type MCPServer struct { @@ -15,6 +16,15 @@ type MCPServer struct { AllowedTools []string `json:"allowed_tools,omitempty"` } +type WebSearchConfig struct { + AllowedDomains []string `json:"allowed_domains,omitempty"` + BlockedDomains []string `json:"blocked_domains,omitempty"` + FetchAllowedDomains []string `json:"fetch_allowed_domains,omitempty"` + MaxUses int `json:"max_uses,omitempty"` + Fetch bool `json:"fetch,omitempty"` + MaxContentTokens int `json:"max_content_tokens,omitempty"` +} + const ( ThinkingModeAdaptive = "adaptive" ThinkingModeDisabled = "disabled" @@ -22,6 +32,28 @@ const ( ThinkingDisplayOmitted = "omitted" ) +// maxDebounceMs bounds debounce_ms. Beyond this the bot reads as unresponsive +// rather than deliberate, and the coalesced turn drifts far enough from the +// user's last message that the reply feels stale. +const maxDebounceMs = 30000 + +// DebounceWindow is the quiet period an intake buffer waits before dispatching a +// coalesced turn. Zero disables debouncing entirely (the default), matching the +// opt-in behavior of comparable gateways. +func (c *BotConfig) DebounceWindow() time.Duration { + if c.DebounceMs <= 0 { + return 0 + } + return time.Duration(c.DebounceMs) * time.Millisecond +} + +// CacheHistoryEnabled reports whether a cache_control breakpoint should be placed +// on the trailing conversation block in addition to the system prompt. Defaults +// to true; set "cache_history": false to opt out. +func (c *BotConfig) CacheHistoryEnabled() bool { + return c.CacheHistory == nil || *c.CacheHistory +} + type BotConfig struct { ID string `json:"id"` TelegramToken string `json:"telegram_token"` @@ -34,6 +66,8 @@ type BotConfig struct { MaxTokens int `json:"max_tokens,omitempty"` Thinking string `json:"thinking,omitempty"` ThinkingDisplay string `json:"thinking_display,omitempty"` + DebounceMs int `json:"debounce_ms,omitempty"` + CacheHistory *bool `json:"cache_history,omitempty"` SystemPrompts map[string]string `json:"system_prompts"` Active bool `json:"active"` OwnerTelegramID int64 `json:"owner_telegram_id"` @@ -43,6 +77,7 @@ type BotConfig struct { ElevenLabsModel string `json:"elevenlabs_model"` DebugScreening bool `json:"debug_screening"` MCPServers []MCPServer `json:"mcp_servers,omitempty"` + WebSearch *WebSearchConfig `json:"web_search,omitempty"` ConfigFilePath string `json:"-"` } @@ -107,10 +142,7 @@ func loadAllConfigs(dir string) ([]BotConfig, error) { continue } - if config.Thinking == ThinkingModeAdaptive && config.MaxTokens > 0 && config.MaxTokens < 4000 { - InfoLogger.Printf("[%s] thinking=adaptive with max_tokens=%d: thinking tokens count toward max_tokens; consider >= 4000", - config.ID, config.MaxTokens) - } + logConfigAdvisories(&config) config.ConfigFilePath = validPath configs = append(configs, config) @@ -124,6 +156,41 @@ func loadAllConfigs(dir string) ([]BotConfig, error) { return configs, nil } +// logConfigAdvisories emits non-fatal boot-time notes about settings that are +// valid but likely to surprise: silently ineffective, or costlier than intended. +func logConfigAdvisories(config *BotConfig) { + if config.Thinking == ThinkingModeAdaptive && config.MaxTokens > 0 && config.MaxTokens < 4000 { + InfoLogger.Printf("[%s] thinking=adaptive with max_tokens=%d: thinking tokens count toward max_tokens; consider >= 4000", + config.ID, config.MaxTokens) + } + + if config.DebounceMs > 0 { + InfoLogger.Printf("[%s] intake debounce enabled: coalescing rapid text messages over a %dms quiet window", + config.ID, config.DebounceMs) + } else { + InfoLogger.Printf("[%s] intake debounce disabled: every message dispatches its own turn (set debounce_ms to coalesce rapid follow-ups)", + config.ID) + } + + if ws := config.WebSearch; ws != nil && len(ws.AllowedDomains) == 0 && len(ws.BlockedDomains) == 0 { + InfoLogger.Printf("[%s] web_search enabled with no allowed_domains/blocked_domains: the model may search the open web", + config.ID) + } + + if ws := config.WebSearch; ws != nil && len(ws.FetchAllowedDomains) > 0 { + if !ws.Fetch { + InfoLogger.Printf("[%s] web_search.fetch_allowed_domains is set but fetch is disabled: it has no effect", + config.ID) + } + for _, d := range ws.FetchAllowedDomains { + if strings.Contains(d, "/") { + InfoLogger.Printf("[%s] web_search.fetch_allowed_domains entry %q includes a path: web_fetch matches host-only, so the whole host is fetchable", + config.ID, d) + } + } + } +} + func validateConfig(config *BotConfig, ids, tokens map[string]bool) error { if config.ID == "" { return fmt.Errorf("missing 'id' field") @@ -168,6 +235,26 @@ func validateConfig(config *BotConfig, ids, tokens map[string]bool) error { return fmt.Errorf("'max_tokens' must be greater than 0 when set") } + if config.DebounceMs < 0 { + return fmt.Errorf("'debounce_ms' must be greater than 0 when set") + } + if config.DebounceMs > maxDebounceMs { + return fmt.Errorf("'debounce_ms' of %d exceeds the maximum of %d (Telegram drops long-idle updates and users read silence as failure)", + config.DebounceMs, maxDebounceMs) + } + + if ws := config.WebSearch; ws != nil { + if len(ws.AllowedDomains) > 0 && len(ws.BlockedDomains) > 0 { + return fmt.Errorf("'web_search' cannot set both allowed_domains and blocked_domains (the API rejects that)") + } + if ws.MaxUses < 0 { + return fmt.Errorf("'web_search.max_uses' must be greater than 0 when set") + } + if ws.MaxContentTokens < 0 { + return fmt.Errorf("'web_search.max_content_tokens' must be greater than 0 when set") + } + } + if config.MessagePerHour <= 0 { return fmt.Errorf("'messages_per_hour' must be greater than 0") } diff --git a/config/default.json b/config/default.json index d3d5e71..2a3c5dc 100644 --- a/config/default.json +++ b/config/default.json @@ -14,6 +14,7 @@ "model": "claude-haiku-4-5", "temperature": 0.7, "max_tokens": 1000, + "debounce_ms": 2500, "debug_screening": false, "system_prompts": { "custom_instructions": "You are Atom, a helpful assistant texting through a limited Telegram interface with a 15-word maximum. Write like texting a friend - use shorthand, skip grammar, use slang/abbreviations. The system cuts off anything longer than 15 words.\n\n- Address the user by their first name, and reply in their preferred language (both are in the conversation context).\n- Use time-appropriate greetings based on the user's local time of day.\n- If a user asks about buying apples, inform them that we don't sell apples.\n- When asked for a joke, tell a clean, family-friendly joke about programming or technology.\n- If someone inquires about our services, explain that we offer AI-powered chatbot solutions.\n- For any questions about pricing, direct users to contact our sales team at sales@example.com.\n- If asked about your capabilities, be honest about what you can and cannot do.\nAlways maintain a friendly and professional tone.", diff --git a/config_test.go b/config_test.go index 6274927..235e71a 100644 --- a/config_test.go +++ b/config_test.go @@ -875,3 +875,84 @@ func TestThinkingConfigLoad(t *testing.T) { t.Errorf("MaxTokens = %d, want 4096", cfg.MaxTokens) } } + +func TestWebSearchConfig(t *testing.T) { + t.Run("absent leaves nil", func(t *testing.T) { + var cfg BotConfig + if err := json.Unmarshal([]byte(`{}`), &cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if cfg.WebSearch != nil { + t.Errorf("WebSearch = %+v, want nil when absent", cfg.WebSearch) + } + }) + + t.Run("loads allowlist", func(t *testing.T) { + jsonData := `{ + "web_search": { + "allowed_domains": ["example.com/hc", "docs.example.com"], + "max_uses": 3, + "fetch": true, + "max_content_tokens": 50000 + } + }` + var cfg BotConfig + if err := json.Unmarshal([]byte(jsonData), &cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if cfg.WebSearch == nil { + t.Fatalf("WebSearch = nil, want populated") + } + if len(cfg.WebSearch.AllowedDomains) != 2 { + t.Errorf("AllowedDomains = %v, want 2 entries", cfg.WebSearch.AllowedDomains) + } + if cfg.WebSearch.MaxUses != 3 { + t.Errorf("MaxUses = %d, want 3", cfg.WebSearch.MaxUses) + } + if !cfg.WebSearch.Fetch { + t.Error("Fetch = false, want true") + } + if cfg.WebSearch.MaxContentTokens != 50000 { + t.Errorf("MaxContentTokens = %d, want 50000", cfg.WebSearch.MaxContentTokens) + } + }) + + cases := []struct { + name string + ws *WebSearchConfig + wantErr string + }{ + {"nil ok", nil, ""}, + {"allowlist ok", &WebSearchConfig{AllowedDomains: []string{"example.com"}, MaxUses: 3}, ""}, + {"blocklist ok", &WebSearchConfig{BlockedDomains: []string{"evil.com"}}, ""}, + {"empty ok", &WebSearchConfig{}, ""}, + {"both allow and block rejected", + &WebSearchConfig{AllowedDomains: []string{"a.com"}, BlockedDomains: []string{"b.com"}}, + "cannot set both allowed_domains and blocked_domains"}, + {"negative max_uses rejected", + &WebSearchConfig{AllowedDomains: []string{"a.com"}, MaxUses: -1}, + "'web_search.max_uses' must be greater than 0"}, + {"negative max_content_tokens rejected", + &WebSearchConfig{AllowedDomains: []string{"a.com"}, MaxContentTokens: -1}, + "'web_search.max_content_tokens' must be greater than 0"}, + } + for i, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := thinkingTestConfig(fmt.Sprintf("bot-web-%d", i)) + cfg.WebSearch = tc.ws + err := validateConfig(&cfg, map[string]bool{}, map[string]bool{}) + if tc.wantErr == "" { + if err != nil { + t.Fatalf("validateConfig(web_search=%+v) = %v, want nil", tc.ws, err) + } + return + } + if err == nil { + t.Fatalf("validateConfig(web_search=%+v) = nil, want error containing %q", tc.ws, tc.wantErr) + } + if !contains(err.Error(), tc.wantErr) { + t.Errorf("error %q does not contain %q", err.Error(), tc.wantErr) + } + }) + } +} diff --git a/go-telegram-bot.exe b/go-telegram-bot.exe index 161f5d4..a391851 100644 Binary files a/go-telegram-bot.exe and b/go-telegram-bot.exe differ diff --git a/handlers.go b/handlers.go index 8ca1e6b..7288cd7 100644 --- a/handlers.go +++ b/handlers.go @@ -28,6 +28,9 @@ func (b *Bot) handleVoiceMessage(ctx context.Context, message *models.Message, u return } + stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping) + defer stopTyping() + transcript, err := b.transcribeVoice(ctx, message.Voice.FileID) if err != nil { ErrorLogger.Printf("Error transcribing voice message from user %d: %v", userID, err) @@ -62,6 +65,12 @@ func (b *Bot) handleVoiceMessage(ctx context.Context, message *models.Message, u return } + // Switch the indicator once the model is done and synthesis begins, so the + // client shows "recording audio" rather than "typing" for a voice reply. + stopTyping() + stopRecording := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionUploadVoice) + defer stopRecording() + audioReader, err := b.generateSpeech(ctx, response) if err != nil { ErrorLogger.Printf("Error generating speech, falling back to text: %v", err) @@ -111,6 +120,11 @@ func (b *Bot) handlePhotoMessage( return } + // Covers the Files API uploads as well as the model turn; on an album this + // is the longest wait in the bot. + stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping) + defer stopTyping() + uploaded := make([]string, len(items)) caption := "" g, gctx := errgroup.WithContext(ctx) @@ -188,6 +202,47 @@ func (b *Bot) handlePhotoMessage( } } +// respondToChat runs one assistant turn against the chat's current memory and +// streams the reply back. Both the immediate path and the debounced flush go +// through here, so a coalesced turn is byte-for-byte the same request as a +// single-message one: the messages were already written to memory at intake, and +// the model simply sees more of them. +func (b *Bot) respondToChat( + ctx context.Context, + chatID, userID int64, + isEmojiOnly bool, + username, firstName, lastName string, + isPremium bool, + languageCode string, + messageTime int, + businessConnectionID string, +) { + stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping) + defer stopTyping() + + chatMemory := b.getOrCreateChatMemory(chatID) + contextMessages := b.prepareContextMessages(chatMemory) + + joined, err := b.getAnthropicResponse( + ctx, chatID, contextMessages, isEmojiOnly, + username, firstName, lastName, isPremium, languageCode, messageTime, + func(seg string) error { + return b.sendOneSegment(ctx, chatID, seg, businessConnectionID) + }, + ) + if err != nil { + ErrorLogger.Printf("Error getting Anthropic response: %v", err) + if sendErr := b.sendResponse(ctx, chatID, b.anthropicErrorResponse(err, userID), businessConnectionID); sendErr != nil { + ErrorLogger.Printf("Error sending response: %v", sendErr) + } + return + } + + if _, storeErr := b.screenOutgoingMessage(chatID, joined); storeErr != nil { + ErrorLogger.Printf("Error recording assistant turn: %v", storeErr) + } +} + func (b *Bot) anthropicErrorResponse(err error, userID int64) string { isElevated := b.hasScope(userID, ScopeModelSet) @@ -268,12 +323,17 @@ func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.U } } + // Media never waits on the text debounce window. Cancelling here does not + // discard the buffered text: those messages are already in chat memory, so + // the turn this media triggers answers them too. if message.MediaGroupID != "" && len(message.Photo) > 0 { + b.cancelIntake(chatID) b.bufferAlbumItem(ctx, message, chatID, userID, username, firstName, lastName, isPremium, languageCode, messageTime, businessConnectionID) return } if len(message.Photo) > 0 { + b.cancelIntake(chatID) if !b.checkRateLimits(userID) { b.sendRateLimitExceededMessage(ctx, chatID, businessConnectionID) return @@ -422,14 +482,14 @@ func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.U } if message.Voice != nil { + b.cancelIntake(chatID) b.handleVoiceMessage(ctx, message, userMsg, chatID, userID, username, firstName, lastName, isPremium, languageCode, messageTime, businessConnectionID) return } - chatMemory := b.getOrCreateChatMemory(chatID) - contextMessages := b.prepareContextMessages(chatMemory) - if message.Sticker != nil { + b.cancelIntake(chatID) + contextMessages := b.prepareContextMessages(b.getOrCreateChatMemory(chatID)) b.handleStickerMessage(ctx, chatID, userMsg, message, contextMessages, businessConnectionID) return } @@ -441,24 +501,18 @@ func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.U isEmojiOnly := isOnlyEmojis(text) - joined, err := b.getAnthropicResponse( - ctx, chatID, contextMessages, isEmojiOnly, - username, firstName, lastName, isPremium, languageCode, messageTime, - func(seg string) error { - return b.sendOneSegment(ctx, chatID, seg, businessConnectionID) - }, - ) - if err != nil { - ErrorLogger.Printf("Error getting Anthropic response: %v", err) - if sendErr := b.sendResponse(ctx, chatID, b.anthropicErrorResponse(err, userID), businessConnectionID); sendErr != nil { - ErrorLogger.Printf("Error sending response: %v", sendErr) - } + // Plain text is the only thing that debounces: it is what users fragment + // across several sends, and it is the only kind whose meaning survives being + // read as one turn. + if b.config.DebounceWindow() > 0 { + b.bufferIntake(ctx, chatID, userID, username, firstName, lastName, + isPremium, languageCode, messageTime, businessConnectionID, isEmojiOnly) return } - if _, storeErr := b.screenOutgoingMessage(chatID, joined); storeErr != nil { - ErrorLogger.Printf("Error recording assistant turn: %v", storeErr) - } + b.respondToChat(ctx, chatID, userID, isEmojiOnly, + username, firstName, lastName, isPremium, languageCode, messageTime, + businessConnectionID) } func (b *Bot) sendRateLimitExceededMessage(ctx context.Context, chatID int64, businessConnectionID string) { @@ -469,7 +523,7 @@ func (b *Bot) sendRateLimitExceededMessage(ctx context.Context, chatID int64, bu func (b *Bot) handleStickerMessage(ctx context.Context, chatID int64, userMessage Message, message *models.Message, contextMessages []anthropic.BetaMessageParam, businessConnectionID string) { - response, err := b.generateStickerResponse(ctx, userMessage, contextMessages) + response, err := b.generateStickerResponse(ctx, userMessage, contextMessages, businessConnectionID) if err != nil { ErrorLogger.Printf("Error generating sticker response: %v", err) if message.Sticker.IsAnimated { @@ -487,7 +541,10 @@ func (b *Bot) handleStickerMessage(ctx context.Context, chatID int64, userMessag } } -func (b *Bot) generateStickerResponse(ctx context.Context, message Message, contextMessages []anthropic.BetaMessageParam) (string, error) { +func (b *Bot) generateStickerResponse(ctx context.Context, message Message, contextMessages []anthropic.BetaMessageParam, businessConnectionID string) (string, error) { + stopTyping := b.startChatAction(ctx, message.ChatID, businessConnectionID, models.ChatActionTyping) + defer stopTyping() + if message.StickerFileID != "" { messageTime := int(message.Timestamp.Unix()) response, err := b.getAnthropicResponse(ctx, message.ChatID, contextMessages, true, message.Username, "", "", false, "", messageTime, nil) @@ -569,6 +626,22 @@ func (b *Bot) clearChatHistory(ctx context.Context, chatID int64, currentUserID return } + // Drop any armed intake buffer for the same chat before clearing memory. + // Otherwise the debounce timer fires moments later and repopulates the chat + // with the very messages that were just deleted — the openclaw/openclaw#51046 + // failure mode, but with a privacy consequence rather than a stray reply. + clearedChatID := chatID + if targetUserID != currentUserID { + clearedChatID = targetChatID + if clearedChatID == 0 { + clearedChatID = targetUserID + } + } + if discarded := b.cancelIntake(clearedChatID); discarded > 0 { + InfoLogger.Printf("[%s] discarded %d buffered message(s) for chat %d on history clear", + b.config.ID, discarded, clearedChatID) + } + b.chatMemoriesMu.Lock() if targetUserID == currentUserID { delete(b.chatMemories, chatID) diff --git a/intake_buffer.go b/intake_buffer.go new file mode 100644 index 0000000..27420ae --- /dev/null +++ b/intake_buffer.go @@ -0,0 +1,129 @@ +package main + +import ( + "context" + "time" +) + +// pendingIntake holds a chat's coalescing window. The buffered message bodies +// are deliberately absent: screenIncomingMessage has already persisted each +// message to the database and to chat memory by the time it is buffered, so the +// flushed turn picks them all up from memory. What is kept here is the metadata +// the turn needs, always refreshed to the most recent message in the batch. +type pendingIntake struct { + chatID, userID int64 + username, firstName, lastName, languageCode string + isPremium bool + messageTime int + businessConnectionID string + allEmojiOnly bool + count int + seq uint64 + timer *time.Timer +} + +// bufferIntake holds a text message for the configured quiet window instead of +// dispatching a turn immediately, resetting the window on each new message. +// Rapid follow-ups therefore produce one reply rather than one per message. +func (b *Bot) bufferIntake( + ctx context.Context, + chatID, userID int64, + username, firstName, lastName string, + isPremium bool, + languageCode string, + messageTime int, + businessConnectionID string, + isEmojiOnly bool, +) { + window := b.config.DebounceWindow() + + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + + pending, exists := b.intakeBuffers[chatID] + if !exists { + b.intakeSeq++ + pending = &pendingIntake{seq: b.intakeSeq, allEmojiOnly: true} + b.intakeBuffers[chatID] = pending + } + + // Reply metadata tracks the most recent message in the batch. + pending.chatID = chatID + pending.userID = userID + pending.username = username + pending.firstName = firstName + pending.lastName = lastName + pending.isPremium = isPremium + pending.languageCode = languageCode + pending.messageTime = messageTime + pending.businessConnectionID = businessConnectionID + pending.allEmojiOnly = pending.allEmojiOnly && isEmojiOnly + pending.count++ + + if pending.timer != nil { + pending.timer.Stop() + } + seq := pending.seq + pending.timer = time.AfterFunc(window, func() { + b.flushIntake(ctx, chatID, seq) + }) +} + +// flushIntake dispatches the coalesced turn for a chat. seq guards against a +// timer that had already fired before its Stop call landed: a stale goroutine +// would otherwise flush a buffer belonging to a later batch. +func (b *Bot) flushIntake(ctx context.Context, chatID int64, seq uint64) { + b.intakeBuffersMu.Lock() + pending, exists := b.intakeBuffers[chatID] + if !exists || pending.seq != seq { + b.intakeBuffersMu.Unlock() + return + } + delete(b.intakeBuffers, chatID) + captured := *pending + b.intakeBuffersMu.Unlock() + + if captured.count > 1 { + InfoLogger.Printf("[%s] intake flush: coalesced %d messages into one turn for chat %d", + b.config.ID, captured.count, chatID) + } + + b.respondToChat( + ctx, chatID, captured.userID, captured.allEmojiOnly, + captured.username, captured.firstName, captured.lastName, + captured.isPremium, captured.languageCode, captured.messageTime, + captured.businessConnectionID, + ) +} + +// cancelIntake drops a chat's pending buffer without dispatching, returning how +// many messages were discarded. +// +// This is the fix for the class of bug in openclaw/openclaw#51046, where a stop +// command aborted the running turn but left the debounce buffer armed, so the +// timer fired afterwards and started the very turn the user had just cancelled. +// Here the stakes are higher than a stray turn: /clear and /clear_hard delete +// chat memory, and a surviving buffer would repopulate it moments later with +// content the user asked to have removed. +func (b *Bot) cancelIntake(chatID int64) int { + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + + pending, exists := b.intakeBuffers[chatID] + if !exists { + return 0 + } + if pending.timer != nil { + pending.timer.Stop() + } + delete(b.intakeBuffers, chatID) + return pending.count +} + +// hasPendingIntake reports whether a chat currently holds a buffered batch. +func (b *Bot) hasPendingIntake(chatID int64) bool { + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + _, exists := b.intakeBuffers[chatID] + return exists +} diff --git a/intake_buffer_test.go b/intake_buffer_test.go new file mode 100644 index 0000000..b91898f --- /dev/null +++ b/intake_buffer_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/go-telegram/bot" + "github.com/go-telegram/bot/models" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// bufferOnly parks a message in the intake buffer without letting the flush run, +// by using a window long enough that no test waits it out. +const bufferOnly = 10 * time.Second + +func bufferText(b *Bot, chatID int64, isEmojiOnly bool) { + b.bufferIntake(context.Background(), chatID, 555, + "tester", "Test", "User", false, "en", int(time.Now().Unix()), "", isEmojiOnly) +} + +func TestBufferIntake_CoalescesIntoSingleTurn(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + // The burst this whole feature exists for: one real question, then filler. + for i := 0; i < 5; i++ { + bufferText(b, 900, false) + } + + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + assert.Len(t, b.intakeBuffers, 1, "rapid messages must share one buffer entry") + assert.Equal(t, 5, b.intakeBuffers[900].count, "all five messages land in the same batch") +} + +// Separate chats must not share a window; one user's burst cannot delay another's. +func TestBufferIntake_IsolatesChats(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + bufferText(b, 910, false) + bufferText(b, 911, false) + bufferText(b, 911, false) + + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + assert.Equal(t, 1, b.intakeBuffers[910].count) + assert.Equal(t, 2, b.intakeBuffers[911].count) +} + +func TestBufferIntake_ResetsWindowAndKeepsLatestMetadata(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + b.bufferIntake(context.Background(), 901, 1, "first", "First", "", false, "en", 1000, "", true) + b.bufferIntake(context.Background(), 901, 2, "second", "Second", "", true, "de", 2000, "biz-42", false) + + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + pending := b.intakeBuffers[901] + require.NotNil(t, pending) + + assert.Equal(t, 2, pending.count) + // OpenClaw semantics: reply metadata follows the most recent message. + assert.Equal(t, "second", pending.username) + assert.Equal(t, int64(2), pending.userID) + assert.Equal(t, "de", pending.languageCode) + assert.Equal(t, 2000, pending.messageTime) + assert.Equal(t, "biz-42", pending.businessConnectionID) + assert.True(t, pending.isPremium) + // One non-emoji message makes the whole coalesced turn non-emoji. + assert.False(t, pending.allEmojiOnly) +} + +func TestBufferIntake_AllEmojiOnlySurvivesWhenEveryMessageIsEmoji(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + bufferText(b, 902, true) + bufferText(b, 902, true) + + b.intakeBuffersMu.Lock() + defer b.intakeBuffersMu.Unlock() + assert.True(t, b.intakeBuffers[902].allEmojiOnly) +} + +func TestCancelIntake_DiscardsPendingBatch(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + bufferText(b, 903, false) + bufferText(b, 903, false) + require.True(t, b.hasPendingIntake(903)) + + assert.Equal(t, 2, b.cancelIntake(903), "cancel reports how many messages it discarded") + assert.False(t, b.hasPendingIntake(903)) + assert.Equal(t, 0, b.cancelIntake(903), "cancelling an empty chat is a no-op") +} + +// A cancelled buffer must never dispatch afterwards. This is the openclaw#51046 +// shape: the timer had already been armed when the cancel landed. +func TestCancelIntake_TimerDoesNotFireAfterCancel(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = 20 + + bufferText(b, 904, false) + b.cancelIntake(904) + + time.Sleep(80 * time.Millisecond) + assert.False(t, b.hasPendingIntake(904), "cancelled buffer must not resurrect") +} + +// A stale timer from a cancelled batch must not flush a newer batch early. +func TestFlushIntake_IgnoresSupersededSequence(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + + bufferText(b, 905, false) + b.intakeBuffersMu.Lock() + staleSeq := b.intakeBuffers[905].seq + b.intakeBuffersMu.Unlock() + + b.cancelIntake(905) + bufferText(b, 905, false) // new batch, new sequence + + // The old timer firing late must not consume the new batch. + b.flushIntake(context.Background(), 905, staleSeq) + assert.True(t, b.hasPendingIntake(905), "superseded flush must leave the newer batch armed") +} + +func TestClearChatHistory_CancelsPendingIntake(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + b.config.DebounceMs = int(bufferOnly / time.Millisecond) + mockTg.SendMessageFunc = func(_ context.Context, _ *bot.SendMessageParams) (*models.Message, error) { + return &models.Message{}, nil + } + + const chatID int64 = 906 + bufferText(b, chatID, false) + require.True(t, b.hasPendingIntake(chatID)) + + b.clearChatHistory(context.Background(), chatID, 123, 0, 0, "", false) + + assert.False(t, b.hasPendingIntake(chatID), + "clearing history must disarm the buffer, or deleted messages get replayed into memory") +} + +func TestDebounceWindow(t *testing.T) { + cases := []struct { + name string + ms int + want time.Duration + }{ + {"unset disables debouncing", 0, 0}, + {"negative disables debouncing", -1, 0}, + {"positive converts to duration", 2500, 2500 * time.Millisecond}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c := BotConfig{DebounceMs: tc.ms} + assert.Equal(t, tc.want, c.DebounceWindow()) + }) + } +} + +func TestCacheHistoryEnabled_DefaultsOn(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + assert.True(t, (&BotConfig{}).CacheHistoryEnabled(), "cache_history defaults to enabled") + + off := false + assert.False(t, (&BotConfig{CacheHistory: &off}).CacheHistoryEnabled()) + + on := true + assert.True(t, (&BotConfig{CacheHistory: &on}).CacheHistoryEnabled()) +} diff --git a/prompt_cache_test.go b/prompt_cache_test.go new file mode 100644 index 0000000..4c9ea4e --- /dev/null +++ b/prompt_cache_test.go @@ -0,0 +1,116 @@ +package main + +import ( + "testing" + + "github.com/anthropics/anthropic-sdk-go" + "github.com/anthropics/anthropic-sdk-go/packages/param" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMarkTrailingCacheBreakpoint(t *testing.T) { + t.Run("marks the final text block of the final turn", func(t *testing.T) { + msgs := []anthropic.BetaMessageParam{ + anthropic.NewBetaUserMessage(anthropic.NewBetaTextBlock("older")), + anthropic.NewBetaUserMessage( + anthropic.NewBetaTextBlock("first"), + anthropic.NewBetaTextBlock("last"), + ), + } + + markTrailingCacheBreakpoint(msgs) + + last := msgs[1].Content[1].OfText + require.NotNil(t, last) + assert.False(t, param.IsOmitted(last.CacheControl), + "the trailing block carries the breakpoint") + + // Everything earlier stays unmarked: one breakpoint, not one per block. + assert.True(t, param.IsOmitted(msgs[1].Content[0].OfText.CacheControl)) + assert.True(t, param.IsOmitted(msgs[0].Content[0].OfText.CacheControl)) + }) + + t.Run("marks a trailing image block", func(t *testing.T) { + msgs := []anthropic.BetaMessageParam{ + anthropic.NewBetaUserMessage( + anthropic.NewBetaImageBlock(anthropic.BetaFileImageSourceParam{FileID: "file_1"}), + ), + } + + markTrailingCacheBreakpoint(msgs) + + require.NotNil(t, msgs[0].Content[0].OfImage) + assert.False(t, param.IsOmitted(msgs[0].Content[0].OfImage.CacheControl)) + }) + + t.Run("tolerates empty input", func(t *testing.T) { + assert.NotPanics(t, func() { markTrailingCacheBreakpoint(nil) }) + assert.NotPanics(t, func() { + markTrailingCacheBreakpoint([]anthropic.BetaMessageParam{{}}) + }) + }) +} + +func TestPrepareContextMessages_CacheHistoryToggle(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, _ := setupBotForTest(t, 123) + memory := &ChatMemory{ + Messages: []Message{ + {IsUser: true, Text: "hello"}, + {IsUser: false, Text: "hi there"}, + }, + Size: 10, + } + + t.Run("enabled by default", func(t *testing.T) { + msgs := b.prepareContextMessages(memory) + require.Len(t, msgs, 2) + assert.False(t, param.IsOmitted(msgs[1].Content[0].OfText.CacheControl)) + }) + + t.Run("opt-out leaves history unmarked", func(t *testing.T) { + off := false + b.config.CacheHistory = &off + defer func() { b.config.CacheHistory = nil }() + + msgs := b.prepareContextMessages(memory) + require.Len(t, msgs, 2) + assert.True(t, param.IsOmitted(msgs[1].Content[0].OfText.CacheControl)) + }) +} + +func TestValidateConfig_DebounceMs(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + base := func(ms int) *BotConfig { + return &BotConfig{ + ID: "b", + TelegramToken: "t", + Model: "claude-sonnet-4-6", + MessagePerHour: 1, + MessagePerDay: 1, + DebounceMs: ms, + } + } + + cases := []struct { + name string + ms int + wantErr bool + }{ + {"omitted is valid", 0, false}, + {"typical chat window is valid", 2500, false}, + {"at the ceiling is valid", maxDebounceMs, false}, + {"negative is rejected", -1, true}, + {"above the ceiling is rejected", maxDebounceMs + 1, true}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := validateConfig(base(tc.ms), map[string]bool{}, map[string]bool{}) + if tc.wantErr { + assert.Error(t, err) + return + } + assert.NoError(t, err) + }) + } +} diff --git a/telegram_client.go b/telegram_client.go index 7ee1791..2f259c0 100644 --- a/telegram_client.go +++ b/telegram_client.go @@ -10,6 +10,7 @@ import ( type TelegramClient interface { SendMessage(ctx context.Context, params *bot.SendMessageParams) (*models.Message, error) SendAudio(ctx context.Context, params *bot.SendAudioParams) (*models.Message, error) + SendChatAction(ctx context.Context, params *bot.SendChatActionParams) (bool, error) SetMyCommands(ctx context.Context, params *bot.SetMyCommandsParams) (bool, error) GetFile(ctx context.Context, params *bot.GetFileParams) (*models.File, error) FileDownloadLink(f *models.File) string diff --git a/telegram_client_mock.go b/telegram_client_mock.go index 5642954..fba27bb 100644 --- a/telegram_client_mock.go +++ b/telegram_client_mock.go @@ -12,6 +12,7 @@ type MockTelegramClient struct { mock.Mock SendMessageFunc func(ctx context.Context, params *bot.SendMessageParams) (*models.Message, error) SendAudioFunc func(ctx context.Context, params *bot.SendAudioParams) (*models.Message, error) + SendChatActionFunc func(ctx context.Context, params *bot.SendChatActionParams) (bool, error) SetMyCommandsFunc func(ctx context.Context, params *bot.SetMyCommandsParams) (bool, error) GetFileFunc func(ctx context.Context, params *bot.GetFileParams) (*models.File, error) FileDownloadLinkFunc func(f *models.File) string @@ -43,6 +44,13 @@ func (m *MockTelegramClient) SendAudio(ctx context.Context, params *bot.SendAudi return nil, nil } +func (m *MockTelegramClient) SendChatAction(ctx context.Context, params *bot.SendChatActionParams) (bool, error) { + if m.SendChatActionFunc != nil { + return m.SendChatActionFunc(ctx, params) + } + return true, nil +} + func (m *MockTelegramClient) GetFile(ctx context.Context, params *bot.GetFileParams) (*models.File, error) { if m.GetFileFunc != nil { return m.GetFileFunc(ctx, params) diff --git a/typing.go b/typing.go new file mode 100644 index 0000000..704d742 --- /dev/null +++ b/typing.go @@ -0,0 +1,65 @@ +package main + +import ( + "context" + "time" + + "github.com/go-telegram/bot" + "github.com/go-telegram/bot/models" +) + +// typingRefreshInterval re-arms the chat action before Telegram expires it. +// The Bot API sets the status "for 5 seconds or less", so anything at or above +// 5s leaves visible gaps. Telegram also clears the status as soon as the bot +// sends a message, so streamed segments naturally interrupt it until the next +// tick; there is no API call to clear it early. +const typingRefreshInterval = 4 * time.Second + +// startChatAction shows a chat action (typing, uploading a photo, recording a +// voice note) and keeps it alive until the returned stop function runs. +// +// The returned function is idempotent and MUST be deferred by the caller. A +// keepalive loop that can outlive its turn is the failure mode behind +// openclaw/openclaw#27177, where the indicator stuck on until the process was +// restarted, so the loop here owns a derived context and exits on the first of: +// stop being called, or the parent context ending. +func (b *Bot) startChatAction( + ctx context.Context, + chatID int64, + businessConnectionID string, + action models.ChatAction, +) (stop func()) { + actionCtx, cancel := context.WithCancel(ctx) + + send := func() { + params := &bot.SendChatActionParams{ + ChatID: chatID, + Action: action, + } + if businessConnectionID != "" { + params.BusinessConnectionID = businessConnectionID + } + if _, err := b.tgBot.SendChatAction(actionCtx, params); err != nil { + // Cosmetic only: a failed indicator must never abort the turn. + InfoLogger.Printf("[%s] chat action %q failed for chat %d: %v", + b.config.ID, action, chatID, err) + } + } + + send() + + go func() { + ticker := time.NewTicker(typingRefreshInterval) + defer ticker.Stop() + for { + select { + case <-actionCtx.Done(): + return + case <-ticker.C: + send() + } + } + }() + + return cancel +} diff --git a/typing_test.go b/typing_test.go new file mode 100644 index 0000000..60e2fcb --- /dev/null +++ b/typing_test.go @@ -0,0 +1,133 @@ +package main + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-telegram/bot" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestStartChatAction_SendsImmediately(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + + var mu sync.Mutex + var got []*bot.SendChatActionParams + mockTg.SendChatActionFunc = func(_ context.Context, p *bot.SendChatActionParams) (bool, error) { + mu.Lock() + defer mu.Unlock() + got = append(got, p) + return true, nil + } + + stop := b.startChatAction(context.Background(), 42, "biz-7", "typing") + stop() + + mu.Lock() + defer mu.Unlock() + require.Len(t, got, 1, "the indicator must show before the slow work starts, not after") + assert.Equal(t, int64(42), got[0].ChatID) + assert.EqualValues(t, "typing", got[0].Action) + assert.Equal(t, "biz-7", got[0].BusinessConnectionID, + "business chats need the connection id or the indicator never renders") +} + +func TestStartChatAction_OmitsEmptyBusinessConnectionID(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + + var mu sync.Mutex + var captured *bot.SendChatActionParams + mockTg.SendChatActionFunc = func(_ context.Context, p *bot.SendChatActionParams) (bool, error) { + mu.Lock() + defer mu.Unlock() + captured = p + return true, nil + } + + stop := b.startChatAction(context.Background(), 42, "", "typing") + stop() + + mu.Lock() + defer mu.Unlock() + require.NotNil(t, captured) + assert.Empty(t, captured.BusinessConnectionID) +} + +// The openclaw#27177 guard: once the turn ends, the keepalive must stop. A loop +// that can re-arm after completion left the indicator stuck on until restart. +func TestStartChatAction_StopHaltsKeepalive(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + + var calls atomic.Int32 + mockTg.SendChatActionFunc = func(_ context.Context, _ *bot.SendChatActionParams) (bool, error) { + calls.Add(1) + return true, nil + } + + stop := b.startChatAction(context.Background(), 42, "", "typing") + stop() + after := calls.Load() + + // Well past a refresh tick had the loop survived. + time.Sleep(50 * time.Millisecond) + assert.Equal(t, after, calls.Load(), "no chat action may be sent after stop") +} + +// Cancelling the parent context must also tear the loop down, so a turn aborted +// upstream cannot leak a goroutine that keeps calling Telegram. +func TestStartChatAction_ParentCancelHaltsKeepalive(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + + var calls atomic.Int32 + mockTg.SendChatActionFunc = func(_ context.Context, _ *bot.SendChatActionParams) (bool, error) { + calls.Add(1) + return true, nil + } + + ctx, cancel := context.WithCancel(context.Background()) + stop := b.startChatAction(ctx, 42, "", "typing") + defer stop() + + cancel() + after := calls.Load() + time.Sleep(50 * time.Millisecond) + assert.Equal(t, after, calls.Load(), "parent cancellation must stop the keepalive") +} + +func TestStartChatAction_StopIsIdempotent(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + mockTg.SendChatActionFunc = func(_ context.Context, _ *bot.SendChatActionParams) (bool, error) { + return true, nil + } + + stop := b.startChatAction(context.Background(), 42, "", "typing") + // The voice path calls stop early and again via defer. + assert.NotPanics(t, func() { + stop() + stop() + }) +} + +// A failed indicator is cosmetic and must never surface as a turn failure. +func TestStartChatAction_SendErrorIsNonFatal(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + b, mockTg := setupBotForTest(t, 123) + mockTg.SendChatActionFunc = func(_ context.Context, _ *bot.SendChatActionParams) (bool, error) { + return false, assert.AnError + } + + assert.NotPanics(t, func() { + stop := b.startChatAction(context.Background(), 42, "", "typing") + stop() + }) +} + +// Telegram clears the status after "5 seconds or less", so the refresh must be +// strictly under that or the indicator visibly drops out mid-turn. +func TestTypingRefreshInterval_UnderTelegramExpiry(t *testing.T) { //NOSONAR go:S100 -- underscore separation is idiomatic in Go test names + assert.Less(t, typingRefreshInterval, 5*time.Second, + "Telegram expires a chat action after at most 5s") +}