diff --git a/gateway/session_authority.py b/gateway/session_authority.py index 81872bb07d..0268bc5152 100644 --- a/gateway/session_authority.py +++ b/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) diff --git a/gateway/session_events.py b/gateway/session_events.py index e3e83e4d4d..5aa8636025 100644 --- a/gateway/session_events.py +++ b/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 diff --git a/gateway/session_hosted_rpc.py b/gateway/session_hosted_rpc.py index 302f75e9e2..22a510603a 100644 --- a/gateway/session_hosted_rpc.py +++ b/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) diff --git a/tests/gateway/test_hosted_room_member_tool_progress.py b/tests/gateway/test_hosted_room_member_tool_progress.py new file mode 100644 index 0000000000..734d561382 --- /dev/null +++ b/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) diff --git a/tui_gateway/hosted_room_driver.py b/tui_gateway/hosted_room_driver.py index 7886789572..905afcfd92 100644 --- a/tui_gateway/hosted_room_driver.py +++ b/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 diff --git a/tui_gateway/hosted_room_member_activity.py b/tui_gateway/hosted_room_member_activity.py index 1aaf120aa3..e6ea6bbd8b 100644 --- a/tui_gateway/hosted_room_member_activity.py +++ b/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: diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index 676bdfeb08..163b108b9d 100644 --- a/tui_gateway/hosted_room_service.py +++ b/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()