feat(gateway): live per-tool status line on Slack

Builds on the salvaged typing_status_text plumbing (PR #62007): instead
of a static 'is thinking...', Slack's assistant status line now updates
live as the agent works — 'is running pytest tests/…', 'is reading
docs/api.md…' — and reverts to the static text between tool calls.

Mechanics:
- agent/display.py: build_status_phrase() derives a <=49-char present-
  tense phrase from the existing _TOOL_VERBS table (+ 'is using <name>'
  for plugin/MCP tools; None for _thinking).
- base adapter: supports_status_text capability flag + set_status_text()
  per-chat store, cleared when the typing loop winds down.
- Slack adapter: send_typing() renders the live phrase when set, falling
  back to typing_status_text then 'is thinking...'.
- gateway/run.py: progress_callback stashes the phrase on tool.started
  and clears on tool.completed. Rendering rides the existing
  _keep_typing refresh cadence — zero additional Slack API calls, no
  rate-limit exposure. Works with tool_progress: off (Slack default);
  the callback is now armed whenever the adapter supports status text.
- display.live_status config (full|verb|off, default full): 'verb' hides
  argument previews for shared/customer-facing channels.

Also fixes a latent crash in the cherry-picked from_dict: malformed
non-dict 'extra' sections broke typing_status_text resolution (uses the
already-coerced extra dict).

Design notes: status text is a side-effect display channel only — never
enters the transcript, no prompt-cache impact. Lifecycle guarantees from
the stuck-status fix family are preserved (per-thread tracking,
clear-on-finish via existing stop_typing paths). Related: #45109
(closed; same direction via lifecycle states), #59010/#51363 (native
task cards — complementary, larger scope).
This commit is contained in:
teknium1 2026-07-18 11:58:31 -07:00 committed by Teknium
parent 16604d59ca
commit d4396797c3
10 changed files with 325 additions and 4 deletions

View file

@ -645,6 +645,52 @@ def verb_drops_preview(tool_name: str) -> bool:
return tool_name in _TOOL_VERBS_NO_PREVIEW
def build_status_phrase(tool_name: str, args: dict | None, max_len: int = 49) -> str | None:
"""Build a short present-tense status phrase for platform status surfaces.
Used by text-rendering "typing" indicators (Slack's
``assistant.threads.setStatus`` line) to show what the agent is doing
right now: ``is running scripts/run_tests.sh`` instead of a static
``is thinking...``. The phrase is phrased to follow the bot's display
name ("Hermes is running …"), so it starts lowercase with "is".
Pass ``args=None`` for a verb-only phrase (``is running``) used when
``display.live_status`` is ``verb`` to keep argument previews out of
shared channels.
Returns None for the ``_thinking`` pseudo-tool and when friendly labels
are disabled (callers fall back to their static default). ``max_len``
caps the total phrase length; Slack truncates its status line around 50
characters, so the default stays just under that.
"""
if not tool_name or tool_name == "_thinking":
return None
if not _friendly_tool_labels:
return None
verb = _TOOL_VERBS.get(tool_name)
if verb:
head = f"is {verb[0].lower()}{verb[1:]}"
else:
# Custom / plugin / MCP tools: generic but still informative.
head = f"is using {tool_name}"
phrase = head
if args and verb and tool_name not in _TOOL_VERBS_NO_PREVIEW:
preview = build_tool_preview(tool_name, args, max_len=None)
if preview:
# Previews can contain newlines (terminal commands); keep the
# status to the first line.
preview = preview.splitlines()[0].strip()
phrase = f"{head}{tool_verb_connector(tool_name)}{preview}"
if len(phrase) > max_len - 1:
phrase = phrase[: max_len - 2].rstrip() + ""
else:
phrase = phrase + ""
return phrase
def build_tool_label(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
"""Build a human-phrased status label for a tool call.

View file

@ -637,7 +637,7 @@ class PlatformConfig:
# into extra); string passthrough, no coercion.
_typing_text = data.get("typing_status_text")
if _typing_text is None:
_typing_text = data.get("extra", {}).get("typing_status_text")
_typing_text = extra.get("typing_status_text")
channel_overrides: Dict[str, ChannelOverride] = {}
raw_overrides = data.get("channel_overrides") or {}

View file

@ -58,6 +58,16 @@ _GLOBAL_DEFAULTS: dict[str, Any] = {
# live, just cleaned up after success so the chat doesn't fill up with
# stale breadcrumbs. Failed runs leave bubbles in place as breadcrumbs.
"cleanup_progress": False,
# Live working-state status on platforms whose typing indicator renders
# text (Slack's assistant status line). Values:
# "full" / true -> verb + argument preview ("is running pytest…")
# "verb" -> verb only ("is running…") — keeps file paths and
# commands out of shared channels
# "off" / false -> static text (typing_status_text or "is thinking...")
# Independent of tool_progress: works even when progress bubbles are off
# (Slack's default), and costs no extra API calls — the existing typing
# refresh cadence just renders different text.
"live_status": "full",
}
# ---------------------------------------------------------------------------
@ -263,6 +273,18 @@ def _normalise(setting: str, value: Any) -> Any:
if isinstance(value, str):
return value.lower() in {"true", "1", "yes", "on"}
return bool(value)
if setting == "live_status":
# Tri-state: "full" (verb + preview), "verb" (verb only), "off".
if value is True:
return "full"
if value is False:
return "off"
val = str(value).strip().lower()
if val in {"true", "1", "yes", "on", "all"}:
return "full"
if val in {"false", "0", "no"}:
return "off"
return val if val in {"full", "verb", "off"} else "full"
if setting == "tool_progress_grouping":
val = str(value).lower()
return val if val in ("accumulate", "separate") else "accumulate"

View file

@ -2314,6 +2314,25 @@ class BasePlatformAdapter(ABC):
# preview (see gateway/run.py progress_callback).
supports_code_blocks: bool = False
# Whether this adapter's typing indicator renders TEXT (a status line
# next to the bot name) rather than a native textless bubble. When True,
# the gateway feeds live per-tool status phrases via set_status_text()
# ("is running pytest…") and send_typing() renders them. Textless
# platforms (Telegram, Discord, Matrix, …) keep the default False and
# never see these calls.
supports_status_text: bool = False
def set_status_text(self, chat_id: str, text: Optional[str]) -> None:
"""Set or clear (``None``) the live working-state phrase for a chat.
Cheap, in-memory only: the next typing refresh renders the new text.
No-op storage on adapters that never read ``_status_text``.
"""
if text:
self._status_text[str(chat_id)] = text
else:
self._status_text.pop(str(chat_id), None)
# Whether this adapter can deliver an ASYNC notification back to the agent
# AFTER a turn ends — i.e. wake a fresh turn to surface a background
# process completion (terminal notify_on_complete / watch_patterns) or a
@ -2464,6 +2483,13 @@ class BasePlatformAdapter(ABC):
# Chats where typing indicator is paused (e.g. during approval waits).
# _keep_typing skips send_typing when the chat_id is in this set.
self._typing_paused: set = set()
# Dynamic working-state status text per chat (chat_id -> phrase).
# Set by the gateway on tool starts ("is running pytest…") and read
# by adapters whose typing indicator renders text (Slack's
# assistant.threads.setStatus). The regular _keep_typing refresh
# cadence picks up changes, so updating this dict costs no extra
# platform API calls. Cleared when the typing loop winds down.
self._status_text: Dict[str, str] = {}
@property
def message_len_fn(self) -> Callable[[str], int]:
@ -3950,6 +3976,7 @@ class BasePlatformAdapter(ABC):
except Exception:
pass
self._typing_paused.discard(chat_id)
self._status_text.pop(str(chat_id), None)
async def _stop_typing_refresh(
self,

View file

@ -18364,6 +18364,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# so each progress line would be sent as a separate message.
from gateway.config import Platform
tool_progress_enabled = progress_mode not in {"off", "log"} and source.platform != Platform.WEBHOOK
# Live working-state status for text-rendering typing indicators
# (Slack's assistant status line). Independent of tool_progress —
# Slack defaults tool_progress off (permanent lines spam channels)
# but the status line is ephemeral, so live status stays useful
# there. Rendering rides the existing _keep_typing refresh: the
# callback only stores a phrase on the adapter, costing zero extra
# platform API calls.
_live_status_mode = resolve_display_setting(
user_config, platform_key, "live_status", "full"
)
_live_status_adapter = self._adapter_for_source(source)
if not getattr(_live_status_adapter, "supports_status_text", False):
_live_status_adapter = None
if _live_status_mode == "off":
_live_status_adapter = None
# "log" mode: tool calls are written to ~/.hermes/logs/tool_calls.log
# instead of the chat (#3459 / #3458). Gateway-only by design.
log_mode_enabled = progress_mode == "log" and source.platform != Platform.WEBHOOK
@ -18468,6 +18483,32 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
def progress_callback(event_type: str, tool_name: str = None, preview: str = None, args: dict = None, **kwargs):
"""Callback invoked by agent on tool lifecycle events."""
# Live status line (Slack's assistant status): stash the current
# tool phrase on the adapter; the _keep_typing refresh renders it
# within a couple of seconds. Handled before every other gate
# because it's independent of progress bubbles and queues (Slack
# keeps tool_progress off by default, but the ephemeral status
# line is always safe). Plain dict write — safe from the agent's
# sync worker thread, no event-loop hop needed.
if (
_live_status_adapter is not None
and _live_status_mode != "off"
and tool_name != "_thinking"
):
try:
if event_type == "tool.started" and tool_name and _run_still_current():
from agent.display import build_status_phrase
_phrase = build_status_phrase(
tool_name,
args if _live_status_mode == "full" else None,
)
_live_status_adapter.set_status_text(source.chat_id, _phrase)
elif event_type == "tool.completed":
# Between tools the model is genuinely "thinking"
# again — revert to the static default.
_live_status_adapter.set_status_text(source.chat_id, None)
except Exception as _ls_err:
logger.debug("live status update failed: %s", _ls_err)
# "log" mode: append tool.started lines to the log queue and stay
# silent in chat. Handled before the progress_queue guard because
# log mode runs without a chat progress queue.
@ -19646,7 +19687,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# None callback — so _thinking scratch bubbles never relayed even
# though the progress queue was created for them.
agent.tool_progress_callback = (
progress_callback if (needs_progress_queue or log_mode_enabled) else None
progress_callback
if (
needs_progress_queue
or log_mode_enabled
or _live_status_adapter is not None
)
else None
)
# Discord voice verbal-ack hook (fires once per turn on first tool
# call; armed only when in a voice channel with the mixer running).

View file

@ -421,6 +421,9 @@ class SlackAdapter(BasePlatformAdapter):
MAX_MESSAGE_LENGTH = 39000 # Slack API allows 40,000 chars; leave margin
supports_code_blocks = True # Slack mrkdwn renders fenced code blocks
# Slack's typing indicator is a text status line (assistant.threads
# .setStatus), so the gateway feeds it live per-tool phrases.
supports_status_text = True
splits_long_messages = True # send() chunks via truncate_message(MAX_MESSAGE_LENGTH)
# Slack blocks typed native slash commands inside threads ("/approve is
# not supported in threads. Sorry!"). The adapter rewrites a leading
@ -1600,11 +1603,15 @@ class SlackAdapter(BasePlatformAdapter):
"team_id": str(team_id) if team_id else "",
}
try:
_status = (
self._status_text.get(str(chat_id))
or getattr(self.config, "typing_status_text", None)
or "is thinking..."
)
await self._get_client(chat_id, team_id=team_id).assistant_threads_setStatus(
channel_id=chat_id,
thread_ts=thread_ts,
status=getattr(self.config, "typing_status_text", None)
or "is thinking...",
status=_status,
)
except Exception as e:
# Silently ignore — may lack assistant:write scope or not be

View file

@ -492,3 +492,62 @@ class TestBuildToolLabel:
for tool_name in _TOOL_VERBS:
label = build_tool_label(tool_name, {"query": "x", "path": "x", "url": "x"})
assert label, f"{tool_name} produced empty label"
class TestBuildStatusPhrase:
"""build_status_phrase — live working-state text for Slack's status line."""
def test_builtin_tool_with_preview(self):
from agent.display import build_status_phrase
phrase = build_status_phrase("terminal", {"command": "pytest tests/"})
assert phrase == "is running pytest tests/…"
def test_search_tool_uses_for_connector(self):
from agent.display import build_status_phrase
phrase = build_status_phrase("web_search", {"query": "slack api limits"})
assert phrase == "is searching the web for slack api limits…"
def test_verb_only_when_args_none(self):
# live_status: "verb" mode passes args=None to suppress previews.
from agent.display import build_status_phrase
assert build_status_phrase("terminal", None) == "is running…"
assert build_status_phrase("read_file", None) == "is reading…"
def test_unknown_tool_generic_phrase(self):
from agent.display import build_status_phrase
phrase = build_status_phrase("my_mcp_tool", {"x": 1})
assert phrase == "is using my_mcp_tool…"
def test_thinking_pseudo_tool_returns_none(self):
from agent.display import build_status_phrase
assert build_status_phrase("_thinking", None) is None
assert build_status_phrase("", None) is None
def test_caps_length_for_slack_status_line(self):
from agent.display import build_status_phrase
phrase = build_status_phrase(
"terminal", {"command": "x" * 300}, max_len=49
)
assert phrase is not None and len(phrase) <= 49
assert phrase.endswith("")
def test_multiline_command_keeps_first_line(self):
from agent.display import build_status_phrase
phrase = build_status_phrase(
"terminal", {"command": "make build\nmake test"}
)
assert phrase is not None
assert "\n" not in phrase
def test_respects_friendly_labels_toggle(self):
from agent.display import build_status_phrase, set_friendly_tool_labels
set_friendly_tool_labels(False)
try:
assert build_status_phrase("terminal", {"command": "ls"}) is None
finally:
set_friendly_tool_labels(True)
def test_no_preview_tools_stay_verb_only(self):
from agent.display import build_status_phrase
phrase = build_status_phrase("skills_list", {"category": "devops"})
assert phrase == "is listing skills…"

View file

@ -607,3 +607,38 @@ class TestReasoningStyle:
config = {"display": {"reasoning_style": "SUBTEXT"}}
assert resolve_display_setting(config, "telegram", "reasoning_style") == "subtext"
class TestLiveStatusSetting:
"""display.live_status — tri-state normalisation + platform overrides."""
def test_default_is_full(self):
from gateway.display_config import resolve_display_setting
assert resolve_display_setting({}, "slack", "live_status") == "full"
def test_bool_and_string_coercion(self):
from gateway.display_config import resolve_display_setting
for raw, expected in [
(True, "full"), (False, "off"),
("on", "full"), ("all", "full"),
("no", "off"), ("verb", "verb"),
("bogus", "full"),
]:
config = {"display": {"live_status": raw}}
assert (
resolve_display_setting(config, "slack", "live_status") == expected
), f"raw={raw!r}"
def test_per_platform_override_wins(self):
from gateway.display_config import resolve_display_setting
config = {
"display": {
"live_status": "full",
"platforms": {"slack": {"live_status": "verb"}},
}
}
assert resolve_display_setting(config, "slack", "live_status") == "verb"
assert resolve_display_setting(config, "google_chat", "live_status") == "full"

View file

@ -2211,6 +2211,56 @@ class TestSendTyping:
status="is pouncing… 🐾",
)
@pytest.mark.asyncio
async def test_live_status_text_overrides_default(self, adapter):
# set_status_text() feeds the live per-tool phrase into the next
# typing refresh.
adapter._app.client.assistant_threads_setStatus = AsyncMock()
adapter.set_status_text("C123", "is running pytest…")
await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
adapter._app.client.assistant_threads_setStatus.assert_called_once_with(
channel_id="C123",
thread_ts="parent_ts",
status="is running pytest…",
)
@pytest.mark.asyncio
async def test_live_status_beats_configured_static_text(self):
# Dynamic per-tool phrase wins over typing_status_text while set;
# clearing it falls back to the configured static string.
config = PlatformConfig(
enabled=True, token="xoxb-fake-token",
typing_status_text="is pouncing… 🐾",
)
a = SlackAdapter(config)
a._app = MagicMock()
a._app.client = AsyncMock()
a._app.client.assistant_threads_setStatus = AsyncMock()
a.set_status_text("C123", "is reading docs/api.md…")
await a.send_typing("C123", metadata={"thread_id": "parent_ts"})
assert (
a._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
== "is reading docs/api.md…"
)
a.set_status_text("C123", None)
await a.send_typing("C123", metadata={"thread_id": "parent_ts"})
assert (
a._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
== "is pouncing… 🐾"
)
@pytest.mark.asyncio
async def test_live_status_scoped_per_chat(self, adapter):
# A phrase for one channel must not leak into another channel's
# status line.
adapter._app.client.assistant_threads_setStatus = AsyncMock()
adapter.set_status_text("C_OTHER", "is running pytest…")
await adapter.send_typing("C123", metadata={"thread_id": "parent_ts"})
assert (
adapter._app.client.assistant_threads_setStatus.call_args.kwargs["status"]
== "is thinking..."
)
@pytest.mark.asyncio
async def test_noop_without_thread(self, adapter):
adapter._app.client.assistant_threads_setStatus = AsyncMock()

View file

@ -436,6 +436,34 @@ The same key customizes Google Chat's visible working-state marker message
note that on Google Chat it is a real posted message that gets patched into the
reply, not an ephemeral status.
### Live Status (per-tool)
By default the status line updates **live as the agent works**: instead of a
static `is thinking...`, it shows what the agent is doing right now — `is
running pytest tests/…`, `is reading docs/api.md…`, `is searching the web for
slack api limits…`. Between tool calls it reverts to the static text. This
rides the existing status-refresh cadence, so it makes no additional Slack API
calls, and it works even with `tool_progress: off` (Slack's default) — unlike
progress bubbles, the status line is ephemeral and leaves nothing behind in
the channel.
Control it with `display.live_status` (global or per-platform):
```yaml
display:
platforms:
slack:
# full = verb + argument ("is running pytest…") [default]
# verb = verb only ("is running…") — hides commands/paths,
# useful in shared or customer-facing channels
# off = static text (typing_status_text or "is thinking...")
live_status: full
```
| Key | Default | Description |
|-----|---------|-------------|
| `display.live_status` | `"full"` | Live per-tool status line. `full` shows verb + argument preview; `verb` shows the verb only (keeps file paths and commands out of shared channels); `off` restores the static text. Requires the `assistant:write` scope, same as the static status line. |
### Session Isolation
```yaml