Debounce and caching

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/<code>`, 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.

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 {

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"} {

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)

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

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 [email protected].\n- If asked about your capabilities, be honest about what you can and cannot do.\nAlways maintain a friendly and professional tone.",

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

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)

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

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())
+}

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

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

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)

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

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")
+}