mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
fix(groups): name a missing RPC method as unavailable, and carry group files on the tui_gateway route
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: <m> — 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
This commit is contained in:
1 parent
afe3a9111f
commit
857d224808
14 files changed
+349
-70
No files matched your search
@@ -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<string, unknown>
|
||||
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',
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
|
||||
Reference in new issue
Block a user