From 857d224808810ca2c6145d2962fa81d4b993bb2a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 27 Oct 2026 07:37:16 -0700 Subject: [PATCH] fix(groups): name a missing RPC method as unavailable, and carry group files on the tui_gateway route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Any unknown JSON-RPC method on the tui_gateway route answered "the client and the Hermes backend are out of sync (different versions)", which the Desktop/Webapp toast turns into "update both" even on matching versions. A JSON-RPC client sends no version, so neither route can know the versions differ: both now answer -32601 "unknown method: — not available on this Hermes backend" from one helper (the "unknown method:" prefix clients match on is unchanged). The concrete case was the group composer's groups.attachment.upload, which only the gateway route served, while HostedRoomServerRPC declared accepts_attachments=False so a member turn with a file failed attachments_unsupported_on_route. The tui_gateway route now serves groups.attachment.upload/download over the same room attachment store, and its member transport resolves event-bound bytes per member exactly like the gateway route (bound_submission_payload, shared): documents ride as prompt path lines, images as an in-process-only _hosted_images list handed to the turn as its image paths. Refs #144648 (WA-20) Refs #145184 --- apps/shared/src/gateway-contract.generated.ts | 90 ++++++----- apps/shared/src/gateway-contract.openrpc.json | 36 +++++ gateway/session_controls.py | 3 +- gateway/session_hosted_attachments.py | 29 +++- tests/tui_gateway/test_groups_route_parity.py | 153 ++++++++++++++++++ tui_gateway/contracts/canonical.py | 15 +- tui_gateway/contracts/prompt_voice.py | 1 + tui_gateway/contracts/registry.py | 6 + tui_gateway/hosted_room_server_rpc.py | 34 +++- tui_gateway/hosted_room_service.py | 2 +- tui_gateway/methods_groups.py | 24 ++- tui_gateway/methods_prompt.py | 21 ++- tui_gateway/rpc_dispatch.py | 3 +- tui_gateway/ws.py | 2 +- 14 files changed, 349 insertions(+), 70 deletions(-) create mode 100644 tests/tui_gateway/test_groups_route_parity.py diff --git a/apps/shared/src/gateway-contract.generated.ts b/apps/shared/src/gateway-contract.generated.ts index b5bd758c6b..fefb528d1c 100644 --- a/apps/shared/src/gateway-contract.generated.ts +++ b/apps/shared/src/gateway-contract.generated.ts @@ -4556,6 +4556,48 @@ export interface PluginCatalogLookupEntry { subdir: string tier?: string } +export interface GroupsAttachmentUploadParams { + profile?: string | null + room_id: string + upload_id: string + kind: string + name: string + mime: string + data_base64: string +} +export interface RoomAttachment { + attachment_id: string + kind: string + name: string + size: number + mime: string + sha256: string + state: string + created_at: number + idempotent: boolean + event_id?: string | null + [key: string]: unknown +} +export interface GroupsAttachmentDownloadParams { + profile?: string | null + room_id: string + event_id: string + attachment_id: string +} +export interface RoomAttachmentBytes { + attachment_id: string + kind: string + name: string + size: number + mime: string + sha256: string + state: string + created_at: number + idempotent: boolean + event_id?: string | null + data_base64: string + [key: string]: unknown +} export interface CanonicalShellExecParams { session_id: string command: string @@ -4897,48 +4939,6 @@ export interface WorkerPersistParams { } /** The operation's durable receipt (``message_id`` for an append, delegation results, …). */ export type WorkerPersistResult = Record -export interface GroupsAttachmentUploadParams { - profile?: string | null - room_id: string - upload_id: string - kind: string - name: string - mime: string - data_base64: string -} -export interface RoomAttachment { - attachment_id: string - kind: string - name: string - size: number - mime: string - sha256: string - state: string - created_at: number - idempotent: boolean - event_id?: string | null - [key: string]: unknown -} -export interface GroupsAttachmentDownloadParams { - profile?: string | null - room_id: string - event_id: string - attachment_id: string -} -export interface RoomAttachmentBytes { - attachment_id: string - kind: string - name: string - size: number - mime: string - sha256: string - state: string - created_at: number - idempotent: boolean - event_id?: string | null - data_base64: string - [key: string]: unknown -} export interface GroupsDiscardParams { profile?: string | null room_id: string @@ -6152,6 +6152,10 @@ export interface RpcMethods { 'gateway.capabilities': { params: PingParams; result: GatewayCapabilitiesResult } /** Resolve one exact pending approval raised by a local or peer room member. */ 'groups.approve': { params: GroupsApproveParams; result: GroupsApproveResult } + /** Read one committed attachment for a live room viewer. */ + 'groups.attachment.download': { params: GroupsAttachmentDownloadParams; result: RoomAttachmentBytes } + /** Store one attachment for a later groups.send manifest (idempotent per upload_id). */ + 'groups.attachment.upload': { params: GroupsAttachmentUploadParams; result: RoomAttachment } /** Describe the hosted-room protocol implemented by this gateway. */ 'groups.capabilities': { params: GroupsCapabilitiesParams; result: GroupsCapabilitiesResult } /** Create a hosted room idempotently; authority is this gateway's stable install identity. */ @@ -6603,6 +6607,8 @@ export const RPC_METHODS = [ 'free_tier.status', 'gateway.capabilities', 'groups.approve', + 'groups.attachment.download', + 'groups.attachment.upload', 'groups.capabilities', 'groups.create', 'groups.demote', diff --git a/apps/shared/src/gateway-contract.openrpc.json b/apps/shared/src/gateway-contract.openrpc.json index 9987c36464..7d750494a5 100644 --- a/apps/shared/src/gateway-contract.openrpc.json +++ b/apps/shared/src/gateway-contract.openrpc.json @@ -1086,6 +1086,42 @@ } } }, + { + "name": "groups.attachment.download", + "summary": "Read one committed attachment for a live room viewer.", + "params": [ + { + "name": "params", + "schema": { + "$ref": "#/components/schemas/GroupsAttachmentDownloadParams" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/RoomAttachmentBytes" + } + } + }, + { + "name": "groups.attachment.upload", + "summary": "Store one attachment for a later groups.send manifest (idempotent per upload_id).", + "params": [ + { + "name": "params", + "schema": { + "$ref": "#/components/schemas/GroupsAttachmentUploadParams" + } + } + ], + "result": { + "name": "result", + "schema": { + "$ref": "#/components/schemas/RoomAttachment" + } + } + }, { "name": "groups.capabilities", "summary": "Describe the hosted-room protocol implemented by this gateway.", diff --git a/gateway/session_controls.py b/gateway/session_controls.py index 680c452ceb..769bb04acc 100644 --- a/gateway/session_controls.py +++ b/gateway/session_controls.py @@ -319,6 +319,7 @@ class AuthorityConnection: ref = SessionRef(self.actor.profile_id, params.get('session_id', '')) from tui_gateway.contracts.registry import ( CANONICAL_METHODS, METHODS, canonical_param_problems, check_params_accepted, check_result, + unknown_method_message, ) contract = CANONICAL_METHODS.get(method) if contract is not None and (problems := canonical_param_problems(contract, params)): @@ -353,7 +354,7 @@ class AuthorityConnection: # 4001 'invalid_params' would read as a bad argument on a method that exists. # An account-owned connector call (the Connectors page) keeps its legacy handler. return {'jsonrpc': '2.0', 'id': rid, 'error': { - 'code': -32601, 'message': f'unknown method: {method}', + 'code': -32601, 'message': unknown_method_message(method), 'data': {'reason': 'unknown_method'}}} else: result = await handlers[method](ref, params) diff --git a/gateway/session_hosted_attachments.py b/gateway/session_hosted_attachments.py index 457b418282..71341fa05d 100644 --- a/gateway/session_hosted_attachments.py +++ b/gateway/session_hosted_attachments.py @@ -23,7 +23,15 @@ def _authorize(service, actor, params, capability): def upload(service, actor, params): - room_id = _authorize(service, actor, params, 'session:submit') + return store_upload(service, _authorize(service, actor, params, 'session:submit'), params) + + +def download(service, actor, params): + return read_download(service, _authorize(service, actor, params, 'session:read'), params) + + +def store_upload(service, room_id, params): + """Retain one composer file for a later ``groups.send`` manifest (both routes).""" encoded = params.get('data_base64') if not isinstance(encoded, str) or len(encoded) > ((MAX_ATTACHMENT_BYTES + 2) // 3) * 4: raise RuntimeStoreError('invalid_params') @@ -36,8 +44,8 @@ def upload(service, actor, params): name=params.get('name'), mime=params.get('mime'), data=data) -def download(service, actor, params): - room_id = _authorize(service, actor, params, 'session:read') +def read_download(service, room_id, params): + """One committed file for a live room viewer (both routes).""" if not params.get('event_id'): raise RuntimeStoreError('invalid_params') gateway_id, epoch = service._owned_authority(room_id) @@ -70,6 +78,14 @@ def append_user_event(service, *, room_id, event_id, payload, gateway_id, epoch) def submission_payload(rpc, prompt, attachments=None): """Resolve only event-bound, member-authorized bytes, never a caller path.""" + return bound_submission_payload( + rpc.authority.db.db_path, room_id=rpc.room_id, member_id=rpc.member_id, prompt=prompt, + attachments=attachments, transferred=getattr(rpc, 'hosted_attachment_data', None)) + + +def bound_submission_payload(db_path, *, room_id, member_id, prompt, attachments=None, transferred=None): + """``{'text', 'attachments'?}`` for one member turn: documents as prompt path lines, images as + owner-local staging paths. Shared by both routes, so a file reaches a member identically.""" if not attachments: return {'text': prompt} from gateway.hosted_room_driver import validate_bound_task_manifest @@ -78,15 +94,14 @@ def submission_payload(rpc, prompt, attachments=None): # Each bound file is captured on its own, so the admission-wide cap is enforced here, # before any member is materialized. validate_media_batch_size(item['size'] for item in manifest) - store = HostedRoomAttachmentStore(rpc.authority.db.db_path) + store = HostedRoomAttachmentStore(db_path) references = [] - transferred = getattr(rpc, 'hosted_attachment_data', None) if transferred is not None and [item for item, data in transferred] != manifest: raise RuntimeStoreError('permission_denied') for index, item in enumerate(manifest): if transferred is None: - saved = store.read(room_id=rpc.room_id, attachment_id=item['attachment_id'], - event_id=item['event_id'], recipient_member_id=rpc.member_id) + saved = store.read(room_id=room_id, attachment_id=item['attachment_id'], + event_id=item['event_id'], recipient_member_id=member_id) if any(saved.attachment[key] != item[key] for key in ('kind', 'name', 'mime', 'size')): raise RuntimeStoreError('permission_denied') data = saved.data diff --git a/tests/tui_gateway/test_groups_route_parity.py b/tests/tui_gateway/test_groups_route_parity.py new file mode 100644 index 0000000000..011df1a38a --- /dev/null +++ b/tests/tui_gateway/test_groups_route_parity.py @@ -0,0 +1,153 @@ +"""Both server routes answer the group composer alike (#145184 points 2 and 4). + +* A method the backend does not serve says so; a JSON-RPC client sends no version, so neither + route may claim the app and backend are on different versions. +* The ``tui_gateway`` route serves ``groups.attachment.upload``/``download`` and carries a member + turn's bound files, as the gateway route does. +""" + +from __future__ import annotations + +import asyncio +import base64 +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from gateway import hosted_rooms +from gateway.hosted_room_attachments import HostedRoomAttachmentStore +from tests.tui_gateway.test_hosted_room_attachment_routes import ( + _admit, + _legacy_runtime, + _run_until_settled_or_error, + _server_module, +) +from tests.tui_gateway.test_hosted_room_driver_runtime import BINDING, PROFILE, ROOM_ID +import tui_gateway.server as srv +from tui_gateway import methods_groups +from tui_gateway.hosted_room_server_rpc import HostedRoomServerRPC + +_PNG = base64.b64decode( + "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==") + + +def _assert_not_available(error: dict, method: str) -> None: + assert error["code"] == -32601, error + assert error["message"] == f"unknown method: {method} — not available on this Hermes backend" + assert "out of sync" not in error["message"] + + +def test_unknown_method_is_named_unavailable_on_both_routes(tmp_path, monkeypatch): + from gateway.session_controls import AuthorityConnection + from hermes_state import SessionDB + + _assert_not_available(srv.handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "groups.no_such_verb", "params": {}})["error"], + "groups.no_such_verb") + + home = tmp_path / "state" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + with SessionDB(home / "state.db") as db: + authority = SimpleNamespace(profile_id=str(home), instance_id="owner", db=db, events={}, sessions={}) + connection = AuthorityConnection(authority, object(), {"user_id": "reader"}) + reply = asyncio.run(connection.dispatch({"id": 2, "method": "groups.no_such_verb", "params": {}})) + _assert_not_available(reply["error"], "groups.no_such_verb") + + +@pytest.fixture +def home(tmp_path, monkeypatch): + class DurableRunStore: + durable = True + + path = tmp_path / ".hermes" + path.mkdir() + (path / "profiles" / "ops").mkdir(parents=True) + (path / "profiles" / "ops" / "config.yaml").write_text("{}\n") + monkeypatch.setenv("HERMES_HOME", str(path)) + monkeypatch.setattr(srv, "_run_idempotency_store", DurableRunStore(), raising=False) + methods_groups.stop_hosted_room_service(timeout=1.0) + methods_groups.start_hosted_room_service() + yield path + methods_groups.stop_hosted_room_service(timeout=1.0) + + +def _result(envelope): + assert "error" not in envelope, envelope + return envelope["result"] + + +def test_tui_route_uploads_sends_and_downloads_a_group_file(home): + capabilities = _result(srv._methods["groups.capabilities"](1, {})) + assert {"groups.attachment.upload", "groups.attachment.download"} <= set(capabilities["methods"]) + _result(srv._methods["groups.create"](2, { + "room_id": "room-1", "name": "Release room", + "members": [{"member_id": "default", "profile": "default", "handle": "hermes"}, + {"member_id": "ops", "profile": "ops", "handle": "ops"}]})) + + data = b"exact composer bytes" + uploaded = _result(srv._methods["groups.attachment.upload"](3, { + "room_id": "room-1", "upload_id": "upload-1", "kind": "file", "name": "note.txt", + "mime": "text/plain", "data_base64": base64.b64encode(data).decode("ascii")})) + manifest = [{k: uploaded[k] for k in ("attachment_id", "kind", "name", "size", "mime")}] + sent = _result(srv._methods["groups.send"](4, { + "room_id": "room-1", "event_id": "send-1", + "payload": {"text": "@ops read this", "thread_id": "thread-1", "attachments": manifest}})) + + downloaded = _result(srv._methods["groups.attachment.download"](5, { + "room_id": "room-1", "event_id": sent["event"]["event_id"], + "attachment_id": uploaded["attachment_id"]})) + assert base64.b64decode(downloaded["data_base64"]) == data + + +@pytest.fixture +def db(tmp_path: Path, monkeypatch) -> Path: + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "member-home")) + path = tmp_path / "rooms.db" + hosted_rooms.create_room( + path, room_id=ROOM_ID, name="Release room", members=[{"profile": PROFILE, "handle": PROFILE}], + authority_gateway_id=BINDING.gateway_id, now=time.time()) + return path + + +def _bind(db: Path, uploads: list[tuple[str, str, bytes]]) -> list[dict]: + store = HostedRoomAttachmentStore(db) + manifest = [] + for index, (name, mime, data) in enumerate(uploads): + saved = store.put(room_id=ROOM_ID, upload_id=f"upload-{index}", kind="image" if mime.startswith("image/") + else "file", name=name, mime=mime, data=data) + manifest.append({k: saved[k] for k in ("attachment_id", "kind", "name", "size", "mime")}) + store.commit_message(room_id=ROOM_ID, event_id="event-1", manifest=manifest, + recipient_member_ids=["member-ops"]) + return [{**item, "event_id": "event-1"} for item in manifest] + + +def test_tui_member_turn_receives_its_document_and_image(db: Path): + identity = _admit(db, attachments=_bind(db, [("note.txt", "text/plain", b"exact bytes"), + ("shot.png", "image/png", _PNG)])) + server, submits = _server_module() + server._profile_home = lambda _profile: None + runtime = _legacy_runtime(db, HostedRoomServerRPC(server, db_path=db)) + + task = _run_until_settled_or_error(runtime, db, identity) + + assert task["status"] == "settled", runtime.status()["last_error"] + submitted, = submits + document = Path(submitted["text"].split("[Shared attachment] file: ")[1].split("\n")[0]) + assert document.read_bytes() == b"exact bytes" + image, = submitted["_hosted_images"] + assert Path(image).read_bytes() == _PNG + + +def test_tui_member_turn_without_a_file_stays_text_only(db: Path): + identity = _admit(db) + server, submits = _server_module() + runtime = _legacy_runtime(db, HostedRoomServerRPC(server, db_path=db)) + + task = _run_until_settled_or_error(runtime, db, identity) + + assert task["status"] == "settled", runtime.status()["last_error"] + assert submits[0]["text"] == "Read the attached note." + assert "_hosted_images" not in submits[0] diff --git a/tui_gateway/contracts/canonical.py b/tui_gateway/contracts/canonical.py index 393cf24bb4..43a5ce4a33 100644 --- a/tui_gateway/contracts/canonical.py +++ b/tui_gateway/contracts/canonical.py @@ -16,7 +16,7 @@ from .common import McpServerStatus, OpenModel, PlanGauge, ProjectRef from .events import SkinPayload from .groups_bot_relay import RoomParams, RoomTaskReceipt from .prompt_voice import ClarifyLockParams, ClarifyLockResult -from .registry import canonical_method +from .registry import canonical_method, method # ── admissions ──────────────────────────────────────────────────────────────────────────────── @@ -510,7 +510,7 @@ canonical_method("worker.persist", params=WorkerPersistParams, result=WorkerPers doc="Sequence-fenced typed persistence write from an adopted worker.") -# ── hosted rooms (canonical-only verbs) ─────────────────────────────────────────────────────── +# ── hosted rooms ───────────────────────────────────────────────────────────────────────────────── class GroupsAttachmentUploadParams(RoomParams): @@ -534,8 +534,12 @@ class RoomAttachment(OpenModel): event_id: str | None = None +_UPLOAD_DOC = "Store one attachment for a later groups.send manifest (idempotent per upload_id)." +# Both routes serve the room byte verbs with one wire, so a group composer attaches files the same +# way whichever backend owns the room; the canonical table keeps its closed-set key check. canonical_method("groups.attachment.upload", params=GroupsAttachmentUploadParams, result=RoomAttachment, - doc="Store one attachment for a later groups.send manifest (idempotent per upload_id).") + doc=_UPLOAD_DOC) +method("groups.attachment.upload", params=GroupsAttachmentUploadParams, result=RoomAttachment, doc=_UPLOAD_DOC) class GroupsAttachmentDownloadParams(RoomParams): @@ -547,8 +551,11 @@ class RoomAttachmentBytes(RoomAttachment): data_base64: str +_DOWNLOAD_DOC = "Read one committed attachment for a live room viewer." canonical_method("groups.attachment.download", params=GroupsAttachmentDownloadParams, - result=RoomAttachmentBytes, doc="Read one committed attachment for a live room viewer.") + result=RoomAttachmentBytes, doc=_DOWNLOAD_DOC) +method("groups.attachment.download", params=GroupsAttachmentDownloadParams, result=RoomAttachmentBytes, + doc=_DOWNLOAD_DOC) class GroupsDiscardParams(RoomParams): diff --git a/tui_gateway/contracts/prompt_voice.py b/tui_gateway/contracts/prompt_voice.py index 1b9ea89616..2f86d68dd3 100644 --- a/tui_gateway/contracts/prompt_voice.py +++ b/tui_gateway/contracts/prompt_voice.py @@ -51,6 +51,7 @@ class PromptSubmitParams(SessionParams): turn_author: JsonValue | None = Field(default=None, exclude=True, alias="_turn_author") hosted_terminal_callback: JsonValue | None = Field( default=None, exclude=True, alias="_hosted_terminal_callback") + hosted_images: JsonValue | None = Field(default=None, exclude=True, alias="_hosted_images") class PromptSubmitStatus(WireEnum): diff --git a/tui_gateway/contracts/registry.py b/tui_gateway/contracts/registry.py index 1dd69d184b..4b8aa8e644 100644 --- a/tui_gateway/contracts/registry.py +++ b/tui_gateway/contracts/registry.py @@ -52,6 +52,12 @@ SERVER_REQUESTS: dict[str, ServerRequestContract] = {} EVENTS: dict[str, EventContract] = {} +def unknown_method_message(method: str) -> str: + """The ``-32601`` text both routes send. A JSON-RPC client sends no version, so a missing + method says only what the backend knows: this backend does not serve it.""" + return f"unknown method: {method} — not available on this Hermes backend" + + def _declare(table: dict, entry) -> None: # Method names may have different envelopes on independently shipped # transports. Duplicates within one transport's catalog are still errors. diff --git a/tui_gateway/hosted_room_server_rpc.py b/tui_gateway/hosted_room_server_rpc.py index 3b4f499aeb..e8896abef1 100644 --- a/tui_gateway/hosted_room_server_rpc.py +++ b/tui_gateway/hosted_room_server_rpc.py @@ -4,9 +4,11 @@ task proof as an in-process-only Python object that JSON clients cannot forge."" from __future__ import annotations +import contextlib import itertools import threading from collections.abc import Mapping, Sequence +from pathlib import Path from types import ModuleType from typing import Any, Callable @@ -27,11 +29,13 @@ class HostedRoomSessionError(RuntimeError): class HostedRoomServerRPC: """Normalize the installed server handlers for :class:`HostedRoomRuntime`.""" - accepts_attachments = False # prompt.submit here has no event-bound file resolution - - def __init__(self, server: ModuleType) -> None: + def __init__(self, server: ModuleType, *, db_path: Path | str | None = None) -> None: self.server = server self._ids = itertools.count(1) + # Event-bound files resolve from the room store, as on the gateway route; without the + # store there is nothing to resolve them from, so the driver refuses the turn by name. + self.db_path = Path(db_path) if db_path is not None else None + self.accepts_attachments = self.db_path is not None def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: envelope = self.server._methods[method](f"hosted-room-{next(self._ids)}", params) @@ -72,11 +76,33 @@ class HostedRoomServerRPC: execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None], member_id: str, attachments: Sequence[Mapping[str, Any]] | None = None, ) -> Mapping[str, Any]: - if attachments: # the driver refuses files before submit on this route + if attachments and self.db_path is None: # the driver refuses files before submit here raise ValueError("attachments_unsupported_on_route: in-process route cannot carry files") + if not attachments: + return self._submit(profile, session_id, prompt, source, task, execution_generation, + on_terminal, member_id, images=()) + from gateway.run import _profile_runtime_scope + from gateway.session_hosted_attachments import bound_submission_payload + from gateway.session_ingress_media import capture_lease + home = self.server._profile_home(profile) + # The member's own caches receive its files (the gateway route's owner scope), and the + # capture lease holds them until prompt.submit has persisted the row that names them. + with capture_lease(): + with (_profile_runtime_scope(home, hydrate_secrets=False) + if home is not None else contextlib.nullcontext()): + inputs = bound_submission_payload( + self.db_path, room_id=task.room_id, member_id=member_id, prompt=prompt, + attachments=list(attachments)) + return self._submit(profile, session_id, inputs["text"], source, task, execution_generation, + on_terminal, member_id, + images=[item["path"] for item in inputs.get("attachments", ())]) + + def _submit(self, profile, session_id, prompt, source, task, execution_generation, on_terminal, + member_id, *, images) -> Mapping[str, Any]: try: return self._call("prompt.submit", { "profile": profile, "session_id": session_id, "text": prompt, "source": source, + **({"_hosted_images": images} if images else {}), "_hosted_task": { "room_id": task.room_id, "task_id": task.task_id, "thread_id": task.thread_id, "turn_id": task.turn_id, "execution_generation": execution_generation, diff --git a/tui_gateway/hosted_room_service.py b/tui_gateway/hosted_room_service.py index a042930f4a..d6d3e59af1 100644 --- a/tui_gateway/hosted_room_service.py +++ b/tui_gateway/hosted_room_service.py @@ -129,7 +129,7 @@ class HostedRoomService: turn_timeout_seconds=_hosted_room_turn_timeout_seconds()) def _make_rpc(self, server): - return HostedRoomServerRPC(server) + return HostedRoomServerRPC(server, db_path=self.db_path) def _load_stored_links(self) -> None: """Rehydrate persisted peer routes; collect per-link errors into one string.""" diff --git a/tui_gateway/methods_groups.py b/tui_gateway/methods_groups.py index e536081737..9cc11e61c6 100644 --- a/tui_gateway/methods_groups.py +++ b/tui_gateway/methods_groups.py @@ -22,7 +22,8 @@ _METHODS = ( "groups.capabilities", "groups.list", "groups.create", "groups.state", "groups.send", "groups.rename", "groups.log", "groups.disband", "groups.replicate", "groups.replica_state", "groups.promote", "groups.demote", "groups.stop", "groups.retry", "groups.approve", - "groups.peer.invite", "groups.peer.revoke", "groups.peer.register") + "groups.peer.invite", "groups.peer.revoke", "groups.peer.register", "groups.attachment.upload", + "groups.attachment.download") LONG_HANDLERS = frozenset(_METHODS) _service_lock = threading.Lock() @@ -402,6 +403,27 @@ def _(rid, params: dict, service) -> dict: "driver_started": True}) +@_room_method( + "groups.attachment.upload", code=5121, room_code=4125, service_code=4123, + service_message=_WORKER_UNAVAILABLE) +def _(rid, params: dict, service) -> dict: + """Retain one composer file for a later ``groups.send`` manifest (same store as the gateway route).""" + from gateway.session_hosted_attachments import store_upload + room_id = params.get("room_id") + service._owned_authority(room_id) + service._room(room_id) + return _ok(rid, store_upload(service, room_id, params)) + + +@_room_method("groups.attachment.download", code=5123, room_code=4126, service_code=4115) +def _(rid, params: dict, service) -> dict: + """Read one committed room file for a viewer of this gateway's room.""" + from gateway.session_hosted_attachments import read_download + room_id = params.get("room_id") + service._room(room_id) + return _ok(rid, read_download(service, room_id, params)) + + @_room_method( "groups.disband", code=5114, room_code=4113, service_code=4123, service_message=_WORKER_UNAVAILABLE) diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index b0957e335c..c7e939d3ab 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -271,12 +271,15 @@ def _typed_stop_phrase_response(rid, text, sid=None): _HOSTED_TASK_FIELDS = {"room_id", "task_id", "thread_id", "turn_id", "execution_generation", "member_id"} -def _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback): - """Validate the hosted-room turn proof carried by an internal submit.""" +def _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback, hosted_images=None): + """Validate the hosted-room turn proof carried by an internal submit. ``hosted_images`` are the + member's event-bound images, already materialized under the image cache by the room worker.""" if session.get("source") != "bot_room": return _err(rid, 4120, "hosted room turns require a bot_room session") valid = ( isinstance(hosted_task, dict) and callable(hosted_terminal_callback) + and (hosted_images is None or (isinstance(hosted_images, list) + and all(isinstance(p, str) and p for p in hosted_images))) and set(hosted_task) == _HOSTED_TASK_FIELDS and all(isinstance(hosted_task.get(f), str) and hosted_task[f] for f in _HOSTED_TASK_FIELDS - {"execution_generation"}) @@ -669,7 +672,8 @@ def _persist_session_row_for_submit(rid, session, text=None, display_kind=None): def _run_after_agent_ready( - rid, sid, session, text, display_kind, display_metadata, hosted_terminal_callback, turn_author=None + rid, sid, session, text, display_kind, display_metadata, hosted_terminal_callback, turn_author=None, + hosted_images=None, ): """Turn thread body: patient wait for a deferred build (a slow build must not eat the accepted in-flight message), then run.""" @@ -700,7 +704,7 @@ def _run_after_agent_ready( return _run_prompt_submit( rid, sid, session, text, display_kind=display_kind, display_metadata=display_metadata, - terminal_callback=hosted_terminal_callback, turn_author=turn_author) + image_paths=hosted_images, terminal_callback=hosted_terminal_callback, turn_author=turn_author) _TRUNCATION_PARAMS = ( @@ -846,9 +850,11 @@ def _(rid, params: dict) -> dict: turn_author = raw_author.author if raw_author is not None else None hosted_task = params.get("_hosted_task") hosted_terminal_callback = params.get("_hosted_terminal_callback") - internal_hosted_submit = hosted_task is not None or hosted_terminal_callback is not None + hosted_images = params.get("_hosted_images") + internal_hosted_submit = ( + hosted_task is not None or hosted_terminal_callback is not None or hosted_images is not None) err = ( - _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback) + _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback, hosted_images) if internal_hosted_submit else _legacy_group_fence_error(rid, session, params)) if err is not None: return err @@ -958,7 +964,8 @@ def _(rid, params: dict) -> dict: _start_agent_build(sid, session) run_thread = threading.Thread( target=lambda: _run_after_agent_ready( - rid, sid, session, text, display_kind, display_metadata, hosted_terminal_callback, turn_author), + rid, sid, session, text, display_kind, display_metadata, hosted_terminal_callback, turn_author, + hosted_images), daemon=True) # Handle lets session.interrupt tell a live turn from a stuck `running` flag. session["_run_thread"] = run_thread diff --git a/tui_gateway/rpc_dispatch.py b/tui_gateway/rpc_dispatch.py index a31b553b07..4197945a90 100644 --- a/tui_gateway/rpc_dispatch.py +++ b/tui_gateway/rpc_dispatch.py @@ -20,8 +20,7 @@ def _handle_admitted_request(req: dict) -> dict | None: return normalized rid, method, params = normalized if not (fn := _methods.get(method)): - return _err(rid, -32601, f"unknown method: {method} — the client and the Hermes backend are out of sync " - "(different versions); run `hermes update` and restart both") + return _err(rid, -32601, _contracts.unknown_method_message(method)) # Test doubles register straight into ``_methods`` without a contract; every production # handler comes through ``register_method`` and therefore has one. contract = _contracts.METHODS.get(method) diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index 12f5139d72..e65a274f70 100644 --- a/tui_gateway/ws.py +++ b/tui_gateway/ws.py @@ -392,7 +392,7 @@ async def _dispatch_request(authority_connection: Any, req: Any, req_method: Any # Session verbs live on the authority; everything else the sidecar still # registers (pet, wake word, connectors) keeps its legacy handler. A legacy # session writer keeps the authority's -32601 (ws_legacy_fallback), as does a - # method neither side knows, which is what the client's version-skew notice keys on. + # method neither side knows, which the client's capability fallbacks key on. resp = await asyncio.to_thread(dispatch_legacy, server, req, transport, actor) return resp