feat(gateway): Slack-native "Thinking Steps" task cards for tool progress (opt-in)

agent/agent_runtime_helpers.py

@@ -1549,7 +1549,17 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
                     or detail.get('content')
                     or detail.get('text')
                 )
-                if summary and summary not in reasoning_parts:
+                # Substring containment, NOT list membership: streamed
+                # responses carry the SAME reasoning twice — once as the
+                # accumulated ``reasoning``/``reasoning_content`` string and
+                # again chunked across reasoning_details thinking blocks.
+                # Each individual block != the accumulated string exactly,
+                # so an equality test appends every block on top of the
+                # full text and the stored reasoning comes out doubled
+                # (observed 2026-07-16: state.db rows byte-identical to
+                # blocks-joined × 2). A block whose text already appears
+                # inside a collected part is a re-delivery, not new content.
+                if summary and not any(summary in part for part in reasoning_parts):
                     reasoning_parts.append(summary)
 
     # Some providers embed reasoning directly inside assistant content

agent/chat_completion_helpers.py

@@ -517,6 +517,14 @@ def interruptible_api_call(agent, api_kwargs: dict):
     the main retry loop can try again with backoff / credential rotation /
     provider fallback.
     """
+    # New response starting — clear the per-response reasoning-delivery
+    # latch so a stale value from a previous (possibly aborted) response
+    # can never suppress this response's reasoning delivery. Consumed by
+    # build_assistant_message; set by _fire_reasoning_delta. Cleared
+    # before the direct-call branch so ALL paths (including cron/inline)
+    # start each response with a clean latch.
+    agent._reasoning_streamed_this_response = False
+
     # Cron and other non-interactive, nested-pool contexts must not spawn the
     # interrupt worker — it wedges before the socket opens on the 2nd+ call
     # (#62151). Run inline instead. See should_use_direct_api_call.
@@ -1265,15 +1273,42 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
         logging.debug(f"Captured reasoning ({len(reasoning_text)} chars): {reasoning_text}")
 
     if reasoning_text and agent.reasoning_callback:
-        # Skip callback when streaming is active — reasoning was already
-        # displayed during the stream via one of two paths:
-        #   (a) _fire_reasoning_delta (structured reasoning_content deltas)
-        #   (b) _stream_delta tag extraction (<think>/<REASONING_SCRATCHPAD>)
-        # When streaming is NOT active, always fire so non-streaming modes
-        # (gateway, batch, quiet) still get reasoning.
-        # Any reasoning that wasn't shown during streaming is caught by the
-        # CLI post-response display fallback (cli.py _reasoning_shown_this_turn).
-        if not agent.stream_delta_callback and not agent._stream_callback:
+        # Deliver reasoning to the callback EXACTLY ONCE per response.
+        # Two independent suppression signals, and BOTH must be checked:
+        #
+        #   (a) the per-response latch set by _fire_reasoning_delta —
+        #       structured reasoning deltas were already delivered
+        #       incrementally to THIS callback during streaming. Gateway
+        #       platforms register reasoning_callback while the *text*
+        #       stream callbacks are None, so signal (b) alone never trips
+        #       there and consumers received every reasoning burst twice:
+        #       once as deltas, once as this full accumulated re-fire
+        #       (observed as 2-4x duplicated sentences on the Slack native
+        #       task cards, 2026-07-05).
+        #
+        #   (b) active text-stream consumers — reasoning that arrives
+        #       inline as <think>/<REASONING_SCRATCHPAD> tags in content is
+        #       extracted and displayed by the CLI's tag-extraction path
+        #       (cli.py _stream_reasoning_delta), which never touches the
+        #       agent latch, so signal (a) alone would re-fire here and
+        #       duplicate the CLI reasoning box. Regression contract:
+        #       tests/cli/test_reasoning_command.py
+        #       (TestReasoningDeltasFiredFlag streaming-active cases).
+        #
+        # When neither signal tripped (non-streaming modes: gateway turns
+        # without reasoning deltas, batch, quiet, streaming-disabled
+        # providers), fire so those consumers still get reasoning exactly
+        # once. The latch is read-and-cleared here and additionally reset
+        # at the start of every API call (interruptible_api_call /
+        # interruptible_streaming_api_call) so a stale value from an
+        # aborted response can never suppress a later delivery.
+        _already_streamed = getattr(agent, "_reasoning_streamed_this_response", False)
+        agent._reasoning_streamed_this_response = False
+        _text_stream_active = bool(
+            getattr(agent, "stream_delta_callback", None)
+            or getattr(agent, "_stream_callback", None)
+        )
+        if not _already_streamed and not _text_stream_active:
             try:
                 agent.reasoning_callback(reasoning_text)
             except Exception:
@@ -2251,6 +2286,12 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
     if agent._interrupt_requested:
         raise InterruptedError("Agent interrupted before streaming API call")
 
+    # New response starting — clear the per-response reasoning-delivery
+    # latch (see the matching comment in interruptible_api_call). The
+    # delegating branches below (direct-call, codex) re-clear via
+    # _interruptible_api_call; clearing here covers the streaming paths.
+    agent._reasoning_streamed_this_response = False
+
     # Cron and other non-interactive, nested-pool contexts deadlock on the
     # spawned worker thread (#62151). They also have no stream consumer, so the
     # deltas this path produces go nowhere. Delegate to the non-streaming entry
@@ -2809,12 +2850,25 @@ def _call_chat_completions(stream_attempt_id: int):
             if hasattr(chunk, "model") and chunk.model:
                 model_name = chunk.model
 
-            # Accumulate reasoning content
+            # Accumulate reasoning content. Providers are not consistent
+            # about what a reasoning "delta" contains: most send true
+            # incremental tokens, but some re-send the ENTIRE accumulated
+            # reasoning as a trailing chunk (cumulative echo) or re-deliver
+            # an overlapping window after an internal reconnect. Naively
+            # appending stores the reasoning doubled (observed 2026-07-05:
+            # state.db rows with byte-identical doubled halves) and fires
+            # duplicated text at reasoning consumers. Normalize here — the
+            # single chokepoint every consumer (trajectory storage,
+            # reasoning_callback, CLI display) sits downstream of.
             reasoning_text = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None)
             if reasoning_text:
-                reasoning_parts.append(reasoning_text)
-                _fire_first_delta()
-                agent._fire_reasoning_delta(reasoning_text)
+                reasoning_text = normalize_reasoning_delta(
+                    "".join(reasoning_parts), reasoning_text
+                )
+                if reasoning_text:
+                    reasoning_parts.append(reasoning_text)
+                    _fire_first_delta()
+                    agent._fire_reasoning_delta(reasoning_text)
 
             # Accumulate text content — fire callback only when no tool calls
             if delta and delta.content:
@@ -3829,6 +3883,70 @@ def _call():
         _reset_stale_streak(agent)
     return result["response"]
 
+# Minimum suffix/prefix overlap (chars) treated as a provider re-delivery
+# rather than legitimate repetition. 24 chars ≈ 6+ tokens — far beyond any
+# plausible legitimate immediate repetition at a token boundary, while
+# reconnect re-deliveries observed in practice overlap by hundreds+ chars.
+_MIN_REASONING_OVERLAP = 24
+
+
+def normalize_reasoning_delta(accumulated: str, delta: str) -> str:
+    """Normalize one incoming reasoning delta against the accumulated text.
+
+    Providers are not consistent about what a reasoning "delta" contains.
+    Three provider misbehaviors are corrected here (all observed in the
+    wild, 2026-07-05 — state.db rows with byte-identical doubled halves):
+
+    1. **Cumulative snapshot** — the chunk re-sends the ENTIRE accumulated
+       reasoning so far (optionally plus new tokens). Detected via
+       ``delta.startswith(accumulated)``; only the new suffix is kept.
+    2. **Exact echo** — the chunk is a byte-identical re-send of the full
+       accumulated text. Dropped (this is case 1 with an empty suffix).
+    3. **Overlapping re-delivery** — after an internal reconnect the
+       provider re-sends a trailing window of already-delivered text
+       followed by new tokens. Detected via a longest suffix(accumulated)/
+       prefix(delta) match; the overlapped head is trimmed.
+
+    A minimum-overlap gate (``_MIN_REASONING_OVERLAP`` chars) protects
+    legitimate repetition: real reasoning frequently repeats short
+    substrings ("the", " so ", word fragments at token boundaries), and a
+    naive containment test (``delta in accumulated``) silently discards
+    such valid tokens. Short overlaps are therefore treated as genuinely
+    new text and appended verbatim — for true incremental token streams
+    (deltas a few chars long) this function is a near-passthrough.
+    Duplicating a few chars in the pathological case is recoverable noise;
+    dropping real tokens is silent data loss, so the gate errs toward
+    appending.
+
+    Returns the (possibly trimmed) text to append, or "" when the delta
+    carries nothing new.
+    """
+    if not delta:
+        return ""
+    if not accumulated:
+        return delta
+    # Case 1+2: cumulative snapshot / exact echo — gated on the accumulated
+    # text being long enough to make a startswith match meaningful. Early in
+    # the stream a SHORT accumulated prefix ("the") is trivially also the
+    # prefix of a legitimate repeated token ("the" again), and an ungated
+    # test silently dropped such tokens (caught by randomized stress tests,
+    # 2026-07-16: streams starting with a repeated word lost the repeat).
+    # The gate is safe for the misbehaviors this function exists to fix:
+    # both observed modes (trailing full-text echo, post-reconnect window
+    # re-delivery) occur late in a stream, when the accumulated text is far
+    # past 24 chars and the gate has long since engaged.
+    if len(accumulated) >= _MIN_REASONING_OVERLAP and delta.startswith(accumulated):
+        return delta[len(accumulated):]
+    # Case 3: overlapping re-delivery — longest suffix of ``accumulated``
+    # that is a prefix of ``delta``, gated to ≥ _MIN_REASONING_OVERLAP so
+    # legitimate short repetitions are never eaten.
+    max_probe = min(len(accumulated), len(delta))
+    for probe in range(max_probe, _MIN_REASONING_OVERLAP - 1, -1):
+        if accumulated.endswith(delta[:probe]):
+            return delta[probe:]
+    return delta
+
+
 # ── Provider fallback ──────────────────────────────────────────────────
 
 
@@ -3837,6 +3955,7 @@ def _call():
     "interruptible_api_call",
     "build_api_kwargs",
     "build_assistant_message",
+    "normalize_reasoning_delta",
     "try_activate_fallback",
     "handle_max_iterations",
     "cleanup_task_resources",

contributors/emails/carnie-bot@openclaw.local

@@ -0,0 +1,2 @@
+menhguin
+# Carnie agent commits — PRs #59009 / #59010

gateway/display_config.py

@@ -33,6 +33,35 @@
 _GLOBAL_DEFAULTS: dict[str, Any] = {
     "tool_progress": "all",
     "tool_progress_grouping": "accumulate",  # "accumulate" = edit one bubble; "separate" = one msg per tool
+    # Opt-in: render tool progress as Slack's native "Thinking Steps" task
+    # cards (chat.startStream/appendStream/stopStream) instead of markdown
+    # progress bubbles. Only takes effect on Slack; every other platform
+    # ignores this key. Off by default — additive, not a behavior change.
+    "tool_progress_native": False,
+    # Card layout for native task cards: "plan" groups all tasks into one
+    # collapsible card (default — reads cleanest, prose lands below it),
+    # "timeline" gives each task its own separate card block, "dense"
+    # collapses consecutive tool calls. Fixed at stream start (Slack limit).
+    "tool_progress_native_mode": "plan",
+    # Native-card tuning knobs (all Slack-only, honored when
+    # tool_progress_native is on). Defaults chosen from live measurement
+    # 2026-07-05; override like any display setting, globally or under
+    # display.platforms.slack.
+    #   rollover_age_s: proactively continue on a fresh streamed message
+    #     after this many seconds. Measured 2026-07-05: Slack kills a
+    #     stream ~306s after startStream even with active appends
+    #     (absolute lifetime, not inactivity) — 240 ≈ 80% with margin.
+    #   rollover_chars: loose size backstop — probed clean past ~62k
+    #     cumulative chars; per-field caps prevent the real msg_too_long
+    #     trigger (single oversized chunk).
+    #   reasoning_chars: cap on the accumulated 💭 reasoning text kept in a
+    #     card's collapsible details. 0 = uncapped (a safety ceiling near
+    #     Slack's 12k markdown_text field limit still applies).
+    #   output_chars: per-tool result preview length on finished cards.
+    "tool_progress_native_rollover_age_s": 240,
+    "tool_progress_native_rollover_chars": 40_000,
+    "tool_progress_native_reasoning_chars": 0,
+    "tool_progress_native_output_chars": 0,
     "show_reasoning": False,
     # How a reasoning/thinking summary is rendered when show_reasoning is on.
     #   "code"      -> 💭 **Reasoning:** + fenced code block (legacy default)
@@ -262,6 +291,7 @@ def _normalise(setting: str, value: Any) -> Any:
         "busy_ack_detail",
         "busy_steer_ack_enabled",
         "thinking_progress",
+        "tool_progress_native",
     }:
         if isinstance(value, str):
             val = value.strip().lower()
@@ -288,12 +318,21 @@ def _normalise(setting: str, value: Any) -> Any:
     if setting == "tool_progress_grouping":
         val = str(value).lower()
         return val if val in ("accumulate", "separate") else "accumulate"
+    if setting == "tool_progress_native_mode":
+        val = str(value).lower()
+        return val if val in ("plan", "timeline", "dense") else "plan"
     if setting == "reasoning_style":
         val = str(value).lower()
         return val if val in ("code", "blockquote", "subtext") else "code"
-    if setting == "tool_preview_length":
+    if setting in {
+        "tool_preview_length",
+        "tool_progress_native_rollover_age_s",
+        "tool_progress_native_rollover_chars",
+        "tool_progress_native_reasoning_chars",
+        "tool_progress_native_output_chars",
+    }:
         try:
-            return int(value)
+            return max(0, int(value))
         except (TypeError, ValueError):
-            return 0
+            return _GLOBAL_DEFAULTS.get(setting, 0)
     return value