fix(groups): a gateway-owned room says which tool a working member is in

groups.state member_activity carried tool/progress_at (1cd4f83) only for the
legacy in-process room path: the map is fed by
tui_gateway.hosted_room_member_activity.project_room_member_activity, which
sees tui_gateway session frames. A gateway-owned (canonical) room runs its
member turns on the gateway's canonical local-session path, whose frames go
SessionAuthority.publish_execution -> the member's SessionEvents stream and
never reach that projection. So the room showed "scout is working... 6s" while
Scout sat in a terminal tool.

Live diagnosis (real Electron + multiplexed gateway, mock inference): one
process (`hermes_cli.main --profile default gateway run`) answered groups.state
(CanonicalHostedRoomService.status), ran the member turn
(mute_notification_reply False, so tool callbacks are wired), published
tool.start/tool.complete name=terminal for the hidden member session, and
answered HostedRoomAuthorityRPC.info for it every active poll. Not a process
split and not a mute gate: the frames just never reach the room's map.

Design: the member side reports, the owner side records, over the info poll
the room driver already makes during a turn (no new polling, and the member
never opens the room owner's database, as session_hosted_transport requires):

- SessionEvents keeps the running execution's open tool calls and last frame
  time, fed by publish_execution after its generation fence (late callbacks
  from another generation move nothing).
- HostedRoomAuthorityRPC.info adds `tool` (newest open call, absent between
  tools) and `progress_at` for its current started hosted task. The same
  answer travels the private owner socket for another local profile's member.
- The driver hands every info answer to a new task_progress callback beside
  pending_action; HostedRoomService records it in the existing bounded map
  only when the info names the same task, and forgets the entry when the
  task's terminal is published. status() merges it as before.

The 1cd4f83 in-process feed stays: the legacy HostedRoomServerRPC path runs
turns in the process answering groups.state, its frames reach the projection
directly, and its info carries no progress fields, so the two feeds never
compete. No wire contract changes: RoomMemberActivity already declares
tool/progress_at and the hosted info verb is internal to the private socket.

gateway/session_authority.py

@@ -1033,6 +1033,7 @@ class SessionAuthority:
             # callback (an agent thread outliving a timed-out turn) never follows the completion.
             if self.executing.get(session_id) != generation:
                 return False
+            live.event_stream.note_progress(generation, event_type, payload)
             delivered = live.event_stream.publish(session_id, payload, event_type=event_type)
             from gateway.session_api_turn import publish_api_event
             observed = publish_api_event(self, session_id, event_type, payload)

gateway/session_events.py

@@ -2,6 +2,7 @@
 from copy import deepcopy
 import logging
 import threading
+import time
 import uuid
 import weakref
 
@@ -29,6 +30,32 @@ class SessionEvents:
         # Set by the authority that attaches transports: retires the subscription of a
         # peer whose bounded backlog overflowed, so it cannot linger looking subscribed.
         self.on_overflow = None
+        # The running execution's open tool calls and its last frame time; a hosted room's info
+        # poll reports them so the room shows which tool a member is in (``progress_for``).
+        self._progress = None
+
+    def note_progress(self, generation, event_type, payload):
+        """Record one execution frame (the caller holds ``lock`` and passed the generation fence)."""
+        progress = self._progress
+        if progress is None or progress['generation'] != generation:
+            progress = self._progress = {'generation': generation, 'tools': {}}
+        payload = payload if isinstance(payload, dict) else {}
+        tool_id = str(payload.get('tool_id') or '')
+        if event_type == 'tool.start':
+            progress['tools'][tool_id] = str(payload.get('name') or '') or 'tool'
+        elif event_type == 'tool.complete':
+            progress['tools'].pop(tool_id, None)
+        progress['at'] = time.time()
+
+    def progress_for(self, generation):
+        """``{"tool", "progress_at"}`` for *generation*'s execution: the newest open tool call
+        (absent between tools) and its last frame time; ``{}`` before its first frame."""
+        with self.lock:
+            progress = self._progress
+            if progress is None or progress['generation'] != generation:
+                return {}
+            tools = list(progress['tools'].values())
+            return {'progress_at': progress['at'], **({'tool': tools[-1]} if tools else {})}
 
     def release(self):
         """The session ended with no viewer left (it may stay resumable): free its replay frames

gateway/session_hosted_rpc.py

@@ -248,6 +248,10 @@ class HostedRoomAuthorityRPC:
                   'task_id': current[1].task_id if current else None,
                   'execution_generation': current[2] if current else None,
                   'status': current[0]['status'] if current else 'idle'}
+        live = self.authority.sessions.get(self.ref.session_id)
+        if current is not None and current[0]['status'] == 'started' and live is not None:
+            # The member's current tool and last frame, for the room's "scout · terminal · 7s".
+            result.update(live.event_stream.progress_for(current[0]['generation']))
         snapshot = await self.authority.attach(self.principal, self.ref, history=False)  # prompts only
         if snapshot.prompts:
             prompt = next((p for p in snapshot.prompts if p.get('kind') == 'approval'), None)

tests/gateway/test_hosted_room_member_tool_progress.py

@@ -0,0 +1,106 @@
+"""A gateway-owned room says which tool a working member is in.
+
+The member's turn runs on the gateway's canonical session path, whose frames never pass through
+tui_gateway's room projection: the member authority's ``info`` reports its live tool, and the room
+driver's existing info poll carries it into ``groups.state`` ``member_activity``.
+"""
+
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+
+from tests.waiting import wait_until
+
+
+def test_member_info_reports_the_running_executions_tool_and_last_frame(hosted_owner):
+    from gateway.hosted_room_driver import TaskIdentity
+    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
+    from hermes_state_runtime import claim_session_input
+    authority, loop, principal, _agent = hosted_owner
+    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='scout', profile='default',
+                                 principal=principal, authorize=lambda *args: True)
+    coords = dict(profile='default', source='bot_room')
+    sid = rpc.create(**coords, title='Group: room')['session_id']
+    rpc.submit(**coords, session_id=sid, prompt='check the logs', task=TaskIdentity('room', 'task', 'thread', 'turn'),
+               execution_generation=1, on_terminal=lambda receipt: None)
+    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
+    authority.executing[sid] = row['generation']  # what the canonical claim stamps before running
+    assert 'tool' not in rpc.info(**coords, session_id=sid)
+
+    authority.publish_execution(sid, row['generation'], 'tool.start', {'tool_id': 'c1', 'name': 'terminal'})
+    info = rpc.info(**coords, session_id=sid)
+    assert (info['task_id'], info['tool']) == ('task', 'terminal') and info['progress_at'] > 0
+
+    # A frame from another execution generation (a late callback) moves nothing.
+    authority.publish_execution(sid, row['generation'] + 1, 'tool.start', {'tool_id': 'c2', 'name': 'web_search'})
+    assert rpc.info(**coords, session_id=sid)['tool'] == 'terminal'
+
+    authority.publish_execution(sid, row['generation'], 'tool.complete', {'tool_id': 'c1', 'name': 'terminal'})
+    info = rpc.info(**coords, session_id=sid)
+    assert 'tool' not in info and info['progress_at'] > 0
+
+
+class _ToolRPC:
+    """Real driver path over a member transport whose ``info`` reports the member's tool, like the
+    gateway's ``HostedRoomAuthorityRPC``; its frames never reach this process's room projection."""
+
+    def __init__(self) -> None:
+        self.task = None
+        self.on_terminal = None
+        self.submitted = threading.Event()
+        self.sessions = {}
+
+    def resolve_exact(self, *, profile, title, source):
+        return self.sessions.get((profile, title))
+
+    def create(self, *, profile, title, source):
+        session = {"session_id": f"{profile}-session", "title": title}
+        self.sessions[(profile, title)] = session
+        return session
+
+    def resume(self, *, profile, session_id, source):
+        return {"session_id": session_id}
+
+    def submit(self, *, task, on_terminal, **_kwargs):
+        self.task, self.on_terminal = task, on_terminal
+        self.submitted.set()
+        return {"accepted": True}
+
+    def history(self, *, profile, session_id, source):
+        return []
+
+    def info(self, *, profile, session_id, source):
+        if self.task is None:
+            return {"active": False, "task_id": None}
+        return {"active": True, "status": "started", "task_id": self.task.task_id, "execution_generation": 1,
+                "tool": "terminal", "progress_at": 1234.5}
+
+
+def test_groups_state_names_the_tool_a_member_transport_reports(tmp_path: Path):
+    from tui_gateway.hosted_room_service import HostedRoomService
+    server = SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock())
+    service = HostedRoomService(server, db_path=tmp_path / "state.db")
+    service.rpc = service.runtime.rpc = _ToolRPC()
+    service.local_profiles = lambda: ("default", "scout")
+    service.create_room(room_id="room-1", name="Release room", members=[
+        {"member_id": "default", "profile": "default", "handle": "hermes"},
+        {"member_id": "scout", "profile": "scout", "handle": "scout"}])
+    service.start()
+    try:
+        service.send(room_id="room-1", event_id="user-1",
+                     payload={"text": "@scout check the build logs", "thread_id": "thread-1"})
+        assert service.rpc.submitted.wait(timeout=5)
+
+        def scout_activity():
+            return next((a for a in service.status("room-1")["member_activity"] if a["member_id"] == "scout"), {})
+        wait_until(lambda: scout_activity().get("tool") == "terminal", timeout=5)
+        assert scout_activity()["progress_at"] == 1234.5
+
+        service.rpc.on_terminal({"status": "settled", "text": "Build logs are clean."})
+        wait_until(lambda: scout_activity() == {}, timeout=5)
+        from tui_gateway.hosted_room_member_activity import task_progress
+        assert task_progress("room-1", service.rpc.task.task_id) == {}
+    finally:
+        service.stop(timeout=5)

tui_gateway/hosted_room_driver.py

@@ -113,6 +113,7 @@ class HostedRoomRuntime:
         prepare_room: Callable[[HostedRoomBinding], None] | None = None,
         publish_terminal: Callable[[HostedRoomBinding, Mapping[str, Any]], None] | None = None,
         pending_action: Callable[[str, str, Mapping[str, Any] | None], None] | None = None,
+        task_progress: Callable[[str, str, Mapping[str, Any]], None] | None = None,
         clock: Callable[[], float] = time.time,
         lease_ttl_seconds: float = 30.0, poll_interval_seconds: float = 5.0,
         active_poll_interval_seconds: float = 0.25, turn_timeout_seconds: float = 1830.0,
@@ -137,7 +138,7 @@ class HostedRoomRuntime:
         self.db_path = Path(db_path)
         self.rpc, self.transport_resolver, self.turn_lock = rpc, transport_resolver, turn_lock
         self.prepare_room, self.publish_terminal = prepare_room, publish_terminal
-        self.pending_action, self.clock = pending_action, clock
+        self.pending_action, self.task_progress, self.clock = pending_action, task_progress, clock
         for name, value in positive.items():
             setattr(self, name, float(value))
         self.max_concurrent_rooms = max_concurrent_rooms
@@ -407,6 +408,9 @@ class HostedRoomRuntime:
 
     def _report_pending_action(
         self, task: Mapping[str, Any], *, session_id: str, info: Mapping[str, Any]) -> None:
+        if self.task_progress is not None:
+            # The member's current tool rides the same info poll as its pending approval.
+            self.task_progress(task["identity"].room_id, task["identity"].task_id, info)
         if self.pending_action is None:
             return
         approval, action = info.get("pending_approval") or info.get("approval"), None

tui_gateway/hosted_room_member_activity.py

@@ -20,8 +20,9 @@ from typing import Any
 
 HOOK_NAME = "on_room_member_activity"
 
-# Per running task: the tool it is in right now and when it last did anything. In-process only (the
-# process running the turn is the one that sees its frames); ``groups.state`` merges it into
+# Per running task: the tool it is in right now and when it last did anything, fed by this process's own
+# room-turn frames (``project_room_member_activity``) or by the member transport's ``info`` answer that
+# the room driver polls during the turn (``record_task_progress``); ``groups.state`` merges it into
 # ``member_activity`` so a room can tell a bot busy in a tool from one gone quiet.
 _PROGRESS_LIMIT = 512
 _progress: OrderedDict[tuple[str, str], dict[str, Any]] = OrderedDict()
@@ -41,6 +42,29 @@ def _note_progress(key: tuple[str, str], kind: str, payload: Mapping[str, Any] |
             _progress.popitem(last=False)
 
 
+def record_task_progress(room_id: str, task_id: str, info: Mapping[str, Any]) -> None:
+    """Adopt the ``tool``/``progress_at`` a member transport's ``info`` reported for this running task.
+
+    A gateway-owned room's member turn publishes its frames on the member authority's own event stream,
+    never through ``project_room_member_activity``; the room driver already polls the member's ``info``
+    during the turn, and this is how that answer reaches ``groups.state``. The member's clock stamps
+    ``progress_at``: the time of its last frame, not of this poll."""
+    progress_at, tool = info.get("progress_at"), info.get("tool")
+    if str(info.get("task_id") or "") != task_id or type(progress_at) not in (int, float):
+        return
+    with _progress_lock:
+        _progress.pop((room_id, task_id), None)
+        _progress[(room_id, task_id)] = {
+            "tool": tool[:128] if isinstance(tool, str) and tool else None, "progress_at": float(progress_at)}
+        while len(_progress) > _PROGRESS_LIMIT:
+            _progress.popitem(last=False)
+
+
+def forget_task_progress(room_id: str, task_id: str) -> None:
+    with _progress_lock:
+        _progress.pop((room_id, task_id), None)
+
+
 def task_progress(room_id: str, task_id: str) -> dict[str, Any]:
     """``{"tool", "progress_at"}`` for a task this process has seen frames from, else ``{}``."""
     with _progress_lock:

tui_gateway/hosted_room_service.py

@@ -23,6 +23,7 @@ from gateway.hosted_room_peer import (
     GatewayRoomCatalog, HostedMemberDispatch, PROTOCOL_VERSION, room_grant_needs_dispatch_refresh)
 from tui_gateway import hosted_room_prepare_stall as prepare_stall
 from tui_gateway.hosted_room_driver import HostedRoomBinding, HostedRoomRuntime
+from tui_gateway.hosted_room_member_activity import forget_task_progress, record_task_progress, task_progress
 from tui_gateway.hosted_room_server_rpc import HostedRoomServerRPC
 from tui_gateway.hosted_room_peer_http import (
     PeerRunsHTTPClient, PeerRunsHTTPError, digest_reauthorization_error)
@@ -51,8 +52,6 @@ def attempt_action(kind: str, task: Mapping[str, Any]) -> dict[str, Any]:
 
 
 def _member_activity(task: Mapping[str, Any]) -> dict[str, Any]:
-    from tui_gateway.hosted_room_member_activity import task_progress
-
     row = {"member_id": task_member_id(task), "status": str(task["status"])}
     if task.get("started_at") is not None:
         row["started_at"] = task["started_at"]
@@ -144,7 +143,7 @@ class HostedRoomService:
             db_path=self.db_path, rooms=self.bindings, rpc=self.rpc,
             transport_resolver=self._resolve_member_transport, turn_lock=self._turn_lock,
             prepare_room=self._prepare_tracked, publish_terminal=self.publish_terminal,
-            pending_action=self._set_pending_action,
+            pending_action=self._set_pending_action, task_progress=record_task_progress,
             poll_interval_seconds=_HOSTED_ROOM_IDLE_FALLBACK_SECONDS,
             active_poll_interval_seconds=_HOSTED_ROOM_ACTIVE_POLL_SECONDS,
             turn_timeout_seconds=_hosted_room_turn_timeout_seconds())
@@ -536,7 +535,8 @@ class HostedRoomService:
             raise
         stalls.clear(binding.room_id)
 
-    def publish_terminal(self, binding: HostedRoomBinding, _task: Mapping[str, Any]) -> None:
+    def publish_terminal(self, binding: HostedRoomBinding, task: Mapping[str, Any]) -> None:
+        forget_task_progress(binding.room_id, task["identity"].task_id)
         self._prepare_tracked(binding)
         self.runtime.wakeup()