fix(slack): send lifecycle values to agents.sessions.setStatus

[KCAYAAI]

Problem

The Agent Sessions status call receives free-text phrases that Slack rejects, and the status does not reflect waiting on a clarify answer.

Change

Sends the documented lifecycle values (working / waiting / done), marks a thread waiting while a clarify prompt is pending, and refreshes an unchanged status periodically so it does not go stale.

Tests

scripts/run_tests.sh tests/gateway/test_slack.py tests/gateway/test_slack_status_update.py → 214 passed, 0 failed

[foo-bender]

Independent mock-based validation of a3a35d9: the documented runner passed the existing two targeted Slack files (213 passed, 1 skipped). A portable regression for free-text → processing and empty clear → active fails on base d0288be and passes on this head (4/4 payload cases). The skip is the real slack-bolt test; this is not live Slack verification.

One P2 cleanup blocker at plugins/platforms/slack/adapter.py:2831–2832: the fire-and-forget pause write can outlive stop_typing. I reproduced both a queued pause that executes after the active clear, and an in-flight suspended request that finishes after the clear. Both leave the last completed status suspended despite _active_status_threads being empty. The generic final-delivery path also invokes pause_typing_for_chat (gateway/platforms/base.py:4525), so this is not limited to clarify/approval waits.

Please order lifecycle writes per workspace/channel/thread and invalidate obsolete turn writes so the final clear wins, including already in-flight writes. A check only at scheduling time would miss the latter case. Event/future-based regression tests reproduce both interleavings without timing sleeps; the added checks currently yield 9 passed / 2 failed.

The core mapping is useful, but I would address this race before approval. No upstream CI checks are reported for the current head.

Portable independent regression tests (test-only patch)

Applies to the tested PR head a3a35d9c59cdb592891e3ca221ccdc32c7c656ed. Save this diff as independent-regressions.patch, apply it with git apply, and run:

scripts/run_tests.sh tests/gateway/test_slack.py -k TestAgentSessionsIndependentValidation

The two cleanup-race tests intentionally fail on the current head; the other nine cases pass. The tests exercise the real adapter methods with mocked Slack clients and deterministic scheduling, not a live Slack workspace.

diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py
index ca893b1e4d..a8d67d5017 100644
--- a/tests/gateway/test_slack.py
+++ b/tests/gateway/test_slack.py
@@ -5757,3 +5757,150 @@ class TestNonConversationalSubtypeAllowlist:
         for event in events:
             accepted = await adapter._prefilter_inbound(event, None)
             assert accepted is not None and accepted[0]["channel"] == "C_FREE", event.get("subtype")
+
+
+class TestAgentSessionsIndependentValidation:
+    """Mock-based lifecycle review coverage; never connects to Slack."""
+
+    def _adapter(self):
+        _slack_mod._AGENT_SESSIONS_SUPPORTED = True
+        a = SlackAdapter(PlatformConfig(enabled=True, token="***"))
+        a._app = MagicMock()
+        a._app.client = AsyncMock()
+        a._app.client.agents_sessions_setStatus = AsyncMock()
+        a._app.client.assistant_threads_setStatus = AsyncMock()
+        return a
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("status,expected", [
+        ("is thinking...", "processing"),
+        ("", "active"),
+        ("suspended", "suspended"),
+        ("closed", "closed"),
+    ])
+    async def test_status_payload_is_lifecycle_enum(self, status, expected):
+        a = self._adapter()
+        await a._set_thread_status("C123", "T1", "100.001", status, "review")
+        a._app.client.agents_sessions_setStatus.assert_awaited_once_with(
+            channel_id="C123", thread_ts="100.001", status=expected)
+        a._app.client.assistant_threads_setStatus.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_failed_status_write_is_retried(self):
+        a = self._adapter()
+        a._app.client.agents_sessions_setStatus.side_effect = [
+            RuntimeError("temporary mock failure"), None]
+        for _ in range(2):
+            await a.send_typing("C123", metadata={"thread_id": "100.001"})
+        assert a._app.client.agents_sessions_setStatus.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_same_state_refreshes_at_sixty_seconds(self, monkeypatch):
+        a = self._adapter()
+        clock = [1000.0]
+        monkeypatch.setattr(_slack_mod.time, "monotonic", lambda: clock[0])
+        metadata = {"thread_id": "100.001"}
+        await a.send_typing("C123", metadata=metadata)
+        clock[0] += 59.0
+        await a.send_typing("C123", metadata=metadata)
+        assert a._app.client.agents_sessions_setStatus.await_count == 1
+        clock[0] += 1.0
+        await a.send_typing("C123", metadata=metadata)
+        assert a._app.client.agents_sessions_setStatus.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_next_turn_in_same_thread_is_not_deduped(self):
+        a = self._adapter()
+        metadata = {"thread_id": "100.001"}
+        await a.send_typing("C123", metadata=metadata)
+        await a.stop_typing("C123", metadata=metadata)
+        await a.send_typing("C123", metadata=metadata)
+        assert [c.kwargs["status"] for c in
+                a._app.client.agents_sessions_setStatus.await_args_list] == [
+                    "processing", "active", "processing"]
+
+    @pytest.mark.asyncio
+    async def test_deduplication_is_workspace_scoped(self):
+        a = self._adapter()
+        first, second = AsyncMock(), AsyncMock()
+        a._team_clients = {"T1": first, "T2": second}
+        for team in ("T1", "T2", "T1", "T2"):
+            await a.send_typing("C123", metadata={
+                "thread_id": "100.001", "team_id": team})
+        first.agents_sessions_setStatus.assert_awaited_once_with(
+            channel_id="C123", thread_ts="100.001", status="processing")
+        second.agents_sessions_setStatus.assert_awaited_once_with(
+            channel_id="C123", thread_ts="100.001", status="processing")
+
+    @pytest.mark.asyncio
+    async def test_typing_disabled_does_not_schedule_lifecycle_writes(self):
+        a = self._adapter()
+        a.config.typing_indicator = False
+        assert a._start_typing_refresh(
+            SimpleNamespace(), asyncio.Event(), {"thread_id": "100.001"}) is None
+        a.pause_typing_for_chat("C123")
+        a.resume_typing_for_chat("C123")
+        a._app.client.agents_sessions_setStatus.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_queued_pause_cannot_overwrite_completed_thread(self, monkeypatch):
+        a = self._adapter()
+        metadata = {"thread_id": "100.001"}
+        await a.send_typing("C123", metadata=metadata)
+        futures = []
+        submit = asyncio.run_coroutine_threadsafe
+
+        def capture(coro, loop):
+            future = submit(coro, loop)
+            futures.append(future)
+            return future
+
+        monkeypatch.setattr(asyncio, "run_coroutine_threadsafe", capture)
+        # The generic final-delivery path also calls pause on the event loop.
+        # An immediately completing mocked clear needs no loop yield, leaving
+        # the scheduled pause behind it. Await the real submitted futures so
+        # the interleaving is deterministic, not a timing/sleep assertion.
+        a.pause_typing_for_chat("C123")
+        await a.stop_typing("C123", metadata=metadata)
+        for future in futures:
+            await asyncio.wrap_future(future)
+        assert not a._active_status_threads
+        assert a._app.client.agents_sessions_setStatus.await_args_list[-1].kwargs[
+            "status"] == "active"
+
+    @pytest.mark.asyncio
+    async def test_inflight_pause_cannot_finish_after_clear(self, monkeypatch):
+        a = self._adapter()
+        metadata = {"thread_id": "100.001"}
+        await a.send_typing("C123", metadata=metadata)
+        entered, release = asyncio.Event(), asyncio.Event()
+        completed = []
+
+        async def delayed_status(**kwargs):
+            if kwargs["status"] == "suspended":
+                entered.set()
+                await release.wait()
+            completed.append(kwargs["status"])
+
+        a._app.client.agents_sessions_setStatus.side_effect = delayed_status
+        futures = []
+        submit = asyncio.run_coroutine_threadsafe
+
+        def capture(coro, loop):
+            future = submit(coro, loop)
+            futures.append(future)
+            return future
+
+        monkeypatch.setattr(asyncio, "run_coroutine_threadsafe", capture)
+        a.pause_typing_for_chat("C123")
+        try:
+            await asyncio.wait_for(entered.wait(), timeout=5)
+            # Model Slack accepting the final clear before the older pause
+            # request finishes. No wall-clock sleep or live API is involved.
+            await a.stop_typing("C123", metadata=metadata)
+        finally:
+            release.set()
+            for future in futures:
+                await asyncio.wrap_future(future)
+        assert not a._active_status_threads
+        assert completed[-1] == "active"