fix(slack): honor Agent Sessions status lifecycle contract

plugins/platforms/slack/adapter.py

@@ -279,15 +279,6 @@ def _sdk_supports_agent_sessions() -> bool:
     return _AGENT_SESSIONS_SUPPORTED
 
 
-def _session_status_method(client: Any):
-    """Return the status setter: Agent Sessions API when available, else legacy."""
-    if _sdk_supports_agent_sessions():
-        method = getattr(client, "agents_sessions_setStatus", None)
-        if method is not None:
-            return method
-    return client.assistant_threads_setStatus
-
-
 def _session_title_method(client: Any):
     """Return the title setter: ``agents.sessions.rename`` when available, else legacy."""
     if _sdk_supports_agent_sessions():
@@ -2520,8 +2511,7 @@ async def _try_finalize_stream(self, chat_id: str, content: str) -> Optional[Sen
         return SendResult(success=True, message_id=ts)
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
-        """Show a thread status via assistant.threads.setStatus.
-        Needs assistant:write or chat:write scope; auto-clears on reply."""
+        """Show a processing session status, or legacy Assistant display text."""
         if self._suppressed_ignored(chat_id, "typing/status in", level=logging.DEBUG):
             return
         if not self._app:
@@ -2564,12 +2554,26 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
 
     async def _set_thread_status(
         self, chat_id: str, team_id: str, thread_ts: str, status: str, fail_label: str) -> None:
-        """``assistant.threads.setStatus`` (empty ``status`` clears); failures are debug-logged."""
+        """Translate legacy display text/clear to the selected API's status contract."""
+        api_method = "session status"
         try:
-            _set_status = _session_status_method(self._get_client(chat_id, team_id=team_id))
+            client = self._get_client(chat_id, team_id=team_id)
+            _set_status = (
+                getattr(client, "agents_sessions_setStatus", None)
+                if _sdk_supports_agent_sessions() else None
+            )
+            if _set_status is not None:
+                api_method = "agents.sessions.setStatus"
+                # Agent Sessions accepts lifecycle enums, not display prose. Clearing typing
+                # leaves the reusable session active; prose never implies suspension/closure.
+                status = "processing" if status else "active"
+            else:
+                api_method = "assistant.threads.setStatus"
+                _set_status = client.assistant_threads_setStatus
             await _set_status(channel_id=chat_id, thread_ts=thread_ts, status=status)
         except Exception as e:
-            logger.debug("[Slack] assistant.threads.setStatus %s: %s", fail_label, e)
+            # SDK exception text/responses can contain credentials or request content.
+            logger.debug("[Slack] %s %s (%s)", api_method, fail_label, type(e).__name__)
 
     @staticmethod
     def _default_status_text(started: Optional[float]) -> str:
@@ -2582,7 +2586,7 @@ def _default_status_text(started: Optional[float]) -> str:
         return f"still working… ({f'{mins}m{secs:02d}s' if mins else f'{secs}s'})"
 
     async def stop_typing(self, chat_id: str, metadata=None) -> None:
-        """Clear the assistant thread status indicator."""
+        """Stop the typing indicator without closing the reusable session."""
         if self._suppressed_ignored(chat_id, "status clear in", level=logging.DEBUG):
             self._active_status_threads.pop(chat_id, None)
             return

tests/gateway/test_slack.py

@@ -5980,47 +5980,88 @@ def _adapter(self):
         a._app.client = AsyncMock()
         return a
 
-    @pytest.mark.asyncio
-    async def test_typing_uses_agent_sessions_when_supported(self):
-        _slack_mod._AGENT_SESSIONS_SUPPORTED = True
+    @pytest.fixture(params=["agent", "legacy-sdk", "missing-method"])
+    def status_transport(self, request, monkeypatch):
+        route = request.param
+        monkeypatch.setattr(_slack_mod, "_AGENT_SESSIONS_SUPPORTED", route != "legacy-sdk")
         a = self._adapter()
-        a._app.client.agents_sessions_setStatus = AsyncMock()
-        a._app.client.assistant_threads_setStatus = AsyncMock()
-        await a.send_typing("C123", metadata={"thread_id": "parent_ts"})
-        a._app.client.agents_sessions_setStatus.assert_called_once_with(
-            channel_id="C123",
-            thread_ts="parent_ts",
-            status="is thinking...",
-        )
-        a._app.client.assistant_threads_setStatus.assert_not_called()
-
+        # Explicit transport attributes: an auto-created mock method hides fallback bugs.
+        client = SimpleNamespace(assistant_threads_setStatus=AsyncMock())
+        if route != "missing-method":
+            client.agents_sessions_setStatus = AsyncMock()
+        a._team_clients = {"T_OTHER": a._app.client, "T_TARGET": client}
+        a._channel_team["C123"] = "T_OTHER"
+        agent_api = route == "agent"
+        setter = client.agents_sessions_setStatus if agent_api else client.assistant_threads_setStatus
+        unused = client.assistant_threads_setStatus if agent_api else getattr(
+            client, "agents_sessions_setStatus", None)
+        return a, setter, unused, agent_api
+
+    @pytest.mark.parametrize("configured,live,elapsed,legacy_status", [
+        pytest.param(None, None, 0, "is thinking...", id="default"),
+        pytest.param("is checking…", None, 0, "is checking…", id="configured"),
+        pytest.param("is checking…", "is reading docs…", 0, "is reading docs…", id="live"),
+        pytest.param(None, None, 123, "still working… (2m03s)", id="elapsed"),
+        pytest.param(None, "Waiting for your answer", 0, "Waiting for your answer", id="wait-prose"),
+        pytest.param(None, "suspended", 0, "suspended", id="lifecycle-word-is-prose"),
+    ])
     @pytest.mark.asyncio
-    async def test_typing_falls_back_to_legacy_without_sdk_support(self):
-        _slack_mod._AGENT_SESSIONS_SUPPORTED = False
-        a = self._adapter()
-        a._app.client.assistant_threads_setStatus = AsyncMock()
-        await a.send_typing("C123", metadata={"thread_id": "parent_ts"})
-        a._app.client.assistant_threads_setStatus.assert_called_once_with(
-            channel_id="C123",
-            thread_ts="parent_ts",
-            status="is thinking...",
-        )
+    async def test_typing_lifecycle_preserves_transport_contract(
+        self, status_transport, monkeypatch, configured, live, elapsed, legacy_status,
+    ):
+        a, setter, unused, agent_api = status_transport
+        a.config.typing_status_text = configured
+        a.set_status_text("C123", live)
+        clock = [1000.0]
+        monkeypatch.setattr(_slack_mod.time, "monotonic", lambda: clock[0])
+        metadata = {"thread_id": "171.000", "message_id": "171.500", "slack_team_id": "T_TARGET"}
+
+        await a.send_typing("C123", metadata=metadata)
+        clock[0] += elapsed
+        await a.send_typing("C123", metadata=metadata)
+        # The tracked workspace must win over the conflicting channel map on clear.
+        await a.stop_typing("C123", metadata={"thread_id": "171.000"})
+        await a.send_typing("C123", metadata=metadata)
+
+        typing_status = "processing" if agent_api else legacy_status
+        fresh_status = "processing" if agent_api else live or configured or "is thinking..."
+        assert setter.await_args_list == [
+            call(channel_id="C123", thread_ts="171.000", status=fresh_status),
+            call(channel_id="C123", thread_ts="171.000", status=typing_status),
+            call(channel_id="C123", thread_ts="171.000", status="active" if agent_api else ""),
+            call(channel_id="C123", thread_ts="171.000", status=fresh_status),
+        ]
+        if unused is not None:
+            unused.assert_not_awaited()
+        assert a._app.client.mock_calls == []
 
+    @pytest.mark.parametrize("operation,fail_label", [
+        ("send_typing", "failed"), ("stop_typing", "clear failed"),
+    ])
     @pytest.mark.asyncio
-    async def test_stop_typing_clears_via_agent_sessions(self):
-        _slack_mod._AGENT_SESSIONS_SUPPORTED = True
-        a = self._adapter()
-        a._app.client.agents_sessions_setStatus = AsyncMock()
-        a._app.client.assistant_threads_setStatus = AsyncMock()
-        await a.send_typing("C123", metadata={"thread_id": "parent_ts"})
-        a._app.client.agents_sessions_setStatus.reset_mock()
-        await a.stop_typing("C123", metadata={"thread_id": "parent_ts"})
-        a._app.client.agents_sessions_setStatus.assert_called_once_with(
-            channel_id="C123",
-            thread_ts="parent_ts",
-            status="",
-        )
-        a._app.client.assistant_threads_setStatus.assert_not_called()
+    async def test_status_transport_failure_is_nonfatal_and_secret_safe(
+        self, status_transport, caplog, operation, fail_label,
+    ):
+        a, setter, unused, agent_api = status_transport
+        secret = "SYNTHETIC_EXCEPTION_SECRET"
+        setter.side_effect = RuntimeError(f"request credentials: {secret}")
+        caplog.set_level("DEBUG", logger=_slack_mod.__name__)
+
+        await getattr(a, operation)(
+            "C123", metadata={"thread_id": "171.000", "slack_team_id": "T_TARGET"},
+        )
+
+        setter.assert_awaited_once()
+        if unused is not None:
+            unused.assert_not_awaited()  # API failures must not retry through the legacy endpoint.
+        assert a._app.client.mock_calls == []
+        records = [record for record in caplog.records if record.name == _slack_mod.__name__]
+        assert records
+        assert secret not in caplog.text
+        method = "agents.sessions.setStatus" if agent_api else "assistant.threads.setStatus"
+        assert any(method in record.getMessage() and fail_label in record.getMessage()
+                   and "RuntimeError" in record.getMessage() for record in records)
+        assert all(record.exc_info is None and record.stack_info is None for record in records)
 
     @pytest.mark.asyncio
     async def test_thread_title_uses_agents_sessions_rename(self):