ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-10-b2-w3' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:14:15 -07:00
parent 73d36e7024 a5e1eef983
commit baef2c5eee
2 changed files with 179 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/hosted_room_discussion.py
+129 -179
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
import hashlib
import re
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from dataclasses import asdict, dataclass
from functools import partial
from typing import Any, Literal
@@ -31,6 +31,7 @@ MAX_USER_TEXT_BYTES = 64 * 1024
MAX_MEMBER_TEXT_BYTES = 64 * 1024
_TRUNCATED_REPLY_NOTICE = "\n\n[Reply truncated. Ask the Bot to share the full result as a file.]"
Payload = Mapping[str, Any]
DecisionStatus = Literal["idle", "task", "settled", "bounded"]
TerminalKind = Literal["settled", "failed", "cancelled", "deferred"]
@@ -40,8 +41,9 @@ _TURN_ID_RE = re.compile(
r"p(?P<position>[0-5])\.s(?P<seen>[1-9][0-9]*)\."
r"m(?P<member>[0-9a-f]{24})$")
_LOCAL_TARGET_FIELDS = frozenset({"kind", "profile"})
_PEER_TARGET_FIELDS = frozenset({"kind", "peer_id", "installation_id", "profile", "capability_digest"})
_TARGET_FIELDS = {
"local": frozenset({"kind", "profile"}),
"peer": frozenset({"kind", "peer_id", "installation_id", "profile", "capability_digest"})}
_REMOTE_MEMBER_FIELDS = frozenset({
"connectionId", "connectionKind", "connectionLabel", "connection_id", "connection_kind", "connection_label",
"remoteSource", "route", "sourceMissing", "sourceReachable", "sourceScoped", "targetProfile", "target_profile"})
@@ -49,12 +51,12 @@ _USER_PAYLOAD_FIELDS = frozenset({"text", "thread_id"})
_TURN_COORDINATE_FIELDS = frozenset(
{"discussion_event_id", "member_id", "member_index", "round_index", "task_id", "thread_id", "turn_id", })
_MEMBER_MESSAGE_FIELDS = _TURN_COORDINATE_FIELDS | {"text"}
_TERMINAL_COMMON_FIELDS = _TURN_COORDINATE_FIELDS | {"seen_through_seq"}
_TERMINAL_EXTRA_FIELDS = {
"turn.settled": frozenset({"message_event_id", "passed"}), "turn.failed": frozenset({"error"}),
"turn.cancelled": frozenset({"reason"}), "turn.deferred": frozenset({"execution_generation", "reason"})}
_TERMINAL_FIELDS = { # kind -> exact payload fields (coordinates + seen_through_seq + per-kind extras)
kind: _TURN_COORDINATE_FIELDS | {"seen_through_seq", *extra} for kind, extra in (
("turn.settled", ("message_event_id", "passed")), ("turn.failed", ("error",)),
("turn.cancelled", ("reason",)), ("turn.deferred", ("execution_generation", "reason")))}
_TERMINAL_OPTIONAL_FIELDS = {"turn.failed": frozenset({"reason_code"})}
_TERMINAL_EVENT_KINDS = frozenset(_TERMINAL_EXTRA_FIELDS)
_TERMINAL_EVENT_KINDS = frozenset(_TERMINAL_FIELDS)
# Gateway-authored control events: kind -> (exact payload fields, identifier fields).
_GATEWAY_EVENT_FIELDS = {
"room.activity": (
@@ -132,11 +134,8 @@ class EventPlan:
authority_epoch: int
def append_kwargs(self, room_id: str) -> dict[str, Any]:
"""Return keyword arguments accepted by ``append_event``."""
return {
"room_id": room_id, "event_id": self.event_id, "kind": self.kind, "actor": dict(self.actor),
"payload": dict(self.payload), "authority_gateway_id": self.authority_gateway_id,
"authority_epoch": self.authority_epoch}
"""Return keyword arguments accepted by ``append_event`` (field order matches the dataclass)."""
return {"room_id": room_id, **asdict(self)}
@dataclass(frozen=True)
@@ -150,12 +149,12 @@ class PublicationPlan:
@dataclass(frozen=True)
class _ValidatedEvent:
raw: Mapping[str, Any]
raw: Payload
seq: int
event_id: str
kind: str
actor: Mapping[str, Any]
payload: Mapping[str, Any]
actor: Payload
payload: Payload
_identifier = partial(common.identifier, error=DiscussionValidationError, max_chars=driver.MAX_IDENTIFIER_CHARS)
@@ -168,15 +167,30 @@ def _positive_int(value: Any, *, label: str) -> int:
def _zero_based_int(value: Any, *, label: str, maximum: int) -> int:
message = f"{label} must be an integer between 0 and {maximum}"
return _bounded_int(value, message=message, high=maximum)
return _bounded_int(value, message=f"{label} must be an integer between 0 and {maximum}", high=maximum)
_text = partial(common.text, error=DiscussionValidationError)
def _peer_target(member: DiscussionMember) -> Mapping[str, Any] | None:
def _peer_id(member: DiscussionMember) -> str | None:
target = member.target
return target if target and target.get("kind") == "peer" else None
return target.get("peer_id") if target and target.get("kind") == "peer" else None
def _member_actor(member: DiscussionMember, *, display_name: bool = True) -> dict[str, Any]:
"""Member actor for the room log; ``connection_id`` (and ``display_name``) only when present."""
actor = {"kind": "member", "id": member.member_id, "profile": member.profile}
if (peer_id := _peer_id(member)) is not None:
actor["connection_id"] = peer_id
if display_name and member.display_name:
actor["display_name"] = member.display_name
return actor
def _all_failure_reasons() -> frozenset[str]:
from tools.bot_failure_reasons import ALL_REASONS
return ALL_REASONS
def validate_user_payload(value: Any) -> dict[str, Any]:
@@ -194,11 +208,9 @@ def _validate_member_target(value: Any, *, profile: str, known_profiles: set[str
if not isinstance(value, Mapping):
raise DiscussionValidationError(f"member {index} target must be an object")
kind = value.get("kind")
if kind not in ("local", "peer"):
if kind not in _TARGET_FIELDS:
raise DiscussionValidationError(f"member {index} target kind must be local or peer")
target = _exact_fields(
value, label=f"member {index} {kind} target",
required=_LOCAL_TARGET_FIELDS if kind == "local" else _PEER_TARGET_FIELDS)
target = _exact_fields(value, label=f"member {index} {kind} target", required=_TARGET_FIELDS[kind])
target_profile = _identifier(target["profile"], label=f"member {index} target profile")
if kind == "local":
if target_profile != profile or profile not in known_profiles:
@@ -215,53 +227,50 @@ def _validate_member_target(value: Any, *, profile: str, known_profiles: set[str
"profile": target_profile, "capability_digest": capability_digest}
def _validate_member(raw: Any, index: int, known_profiles: set[str]) -> DiscussionMember:
if not isinstance(raw, Mapping):
raise DiscussionValidationError(f"member {index} must be an object")
if remote_fields := frozenset(raw) & _REMOTE_MEMBER_FIELDS:
raise DiscussionValidationError(
f"member {index} contains cross-gateway fields: {', '.join(sorted(remote_fields))}")
member = _exact_fields(
raw, label=f"member {index}", required=frozenset({"member_id", "profile", "handle"}),
optional=frozenset({"display_name", "target"}))
member_id, profile, handle = (
_identifier(member[field], label=f"member {index} {label}")
for field, label in (("member_id", "id"), ("profile", "profile"), ("handle", "handle")))
target = _validate_member_target(member.get("target"), profile=profile, known_profiles=known_profiles, index=index)
display_name = member.get("display_name", "")
if not isinstance(display_name, str):
raise DiscussionValidationError(f"member {index} display_name must be a string")
if len(display_name := display_name.strip()) > hosted_rooms.MAX_ACTOR_LABEL_CHARS:
raise DiscussionValidationError(f"member {index} display_name is too long")
return DiscussionMember(member_id, profile, handle, display_name, target)
def validate_roster(value: Any, *, local_profiles: Iterable[str]) -> tuple[DiscussionMember, ...]:
"""Validate a frozen 2-6 member roster of profiles on this gateway."""
if not isinstance(value, list):
raise DiscussionValidationError("members must be a list")
if not MIN_DISCUSSION_MEMBERS <= len(value) <= MAX_DISCUSSION_MEMBERS:
raise DiscussionValidationError(
f"members must contain between {MIN_DISCUSSION_MEMBERS} and "
f"{MAX_DISCUSSION_MEMBERS} entries")
f"members must contain between {MIN_DISCUSSION_MEMBERS} and {MAX_DISCUSSION_MEMBERS} entries")
known_profiles = {_identifier(profile, label="local profile") for profile in local_profiles}
members: list[DiscussionMember] = []
targets: set[str] = set()
handles: set[str] = {"all", "everyone"} # reserved mention handles
member_ids: set[str] = set()
for index, raw in enumerate(value):
if not isinstance(raw, Mapping):
raise DiscussionValidationError(f"member {index} must be an object")
remote_fields = frozenset(raw) & _REMOTE_MEMBER_FIELDS
if remote_fields:
raise DiscussionValidationError(
f"member {index} contains cross-gateway fields: {', '.join(sorted(remote_fields))}")
member = _exact_fields(
raw, label=f"member {index}", required=frozenset({"member_id", "profile", "handle"}),
optional=frozenset({"display_name", "target"}))
member_id = _identifier(member["member_id"], label=f"member {index} id")
profile = _identifier(member["profile"], label=f"member {index} profile")
handle = _identifier(member["handle"], label=f"member {index} handle")
target = _validate_member_target(
member.get("target"), profile=profile, known_profiles=known_profiles, index=index)
display_name = member.get("display_name", "")
if not isinstance(display_name, str):
raise DiscussionValidationError(f"member {index} display_name must be a string")
display_name = display_name.strip()
if len(display_name) > hosted_rooms.MAX_ACTOR_LABEL_CHARS:
raise DiscussionValidationError(f"member {index} display_name is too long")
target_key = compact_json(target, ensure_ascii=False).casefold()
target_message = (
"member profiles must be unique" if target.get("kind") == "local"
else "member targets must be unique")
member = _validate_member(raw, index, known_profiles)
unique = "profiles" if member.target.get("kind") == "local" else "targets"
for key, seen, message in (
(target_key, targets, target_message),
(handle.casefold(), handles, "member handles must be unique and cannot reserve @all or @everyone"),
(member_id.casefold(), member_ids, "member ids must be unique")):
(compact_json(member.target, ensure_ascii=False).casefold(), targets, f"member {unique} must be unique"),
(member.handle.casefold(), handles, "member handles must be unique and cannot reserve @all or @everyone"),
(member.member_id.casefold(), member_ids, "member ids must be unique")):
if key in seen:
raise DiscussionValidationError(message)
seen.add(key)
members.append(DiscussionMember(
member_id=member_id, profile=profile, handle=handle, display_name=display_name, target=target))
members.append(member)
return tuple(members)
@@ -275,14 +284,12 @@ def validate_room(value: Any, *, local_profiles: Iterable[str]) -> DiscussionRoo
name = value.get("name")
if not isinstance(name, str) or not name.strip():
raise DiscussionValidationError("room name must be a non-empty string")
name = name.strip()
if len(name) > hosted_rooms.MAX_ROOM_NAME_CHARS:
if len(name := name.strip()) > hosted_rooms.MAX_ROOM_NAME_CHARS:
raise DiscussionValidationError("room name is too long")
gateway_id = _identifier(value.get("authority_gateway_id"), label="authority_gateway_id")
authority_epoch = _positive_int(value.get("authority_epoch"), label="authority_epoch")
members = validate_roster(value.get("members"), local_profiles=local_profiles)
return DiscussionRoom(
room_id=room_id, name=name, members=members, gateway_id=gateway_id, authority_epoch=authority_epoch)
return DiscussionRoom(room_id, name, members, gateway_id, authority_epoch)
def is_pass_text(value: Any) -> bool:
@@ -337,8 +344,7 @@ def _require_gateway_actor(actor: Mapping[str, Any], room: DiscussionRoom, messa
def _member_by_id(room: DiscussionRoom, member_id: Any) -> DiscussionMember:
normalized = _identifier(member_id, label="member_id")
member = next((m for m in room.members if m.member_id == normalized), None)
if member is None:
if (member := next((m for m in room.members if m.member_id == normalized), None)) is None:
raise DiscussionValidationError(f"unknown Discussion member '{normalized}'")
return member
@@ -355,35 +361,29 @@ def _validate_turn_coordinates(payload: Mapping[str, Any], room: DiscussionRoom)
# Each takes (kind, payload, actor, room) and returns the payload to record.
def _validate_user_event(
kind: str, payload: Mapping[str, Any], actor: Mapping[str, Any], room: DiscussionRoom) -> Mapping[str, Any]:
def _validate_user_event(kind: str, payload: Payload, actor: Payload, room: DiscussionRoom) -> Payload:
payload = validate_user_payload(payload)
if actor.get("kind") != "user":
raise DiscussionValidationError("message.user requires a user actor")
return payload
def _validate_member_message(
kind: str, payload: Mapping[str, Any], actor: Mapping[str, Any], room: DiscussionRoom) -> Mapping[str, Any]:
def _validate_member_message(kind: str, payload: Payload, actor: Payload, room: DiscussionRoom) -> Payload:
_exact_fields(payload, label="message.member payload", required=_MEMBER_MESSAGE_FIELDS)
_validate_turn_coordinates(payload, room)
text = payload.get("text")
if not isinstance(text, str) or not text.strip() or is_pass_text(text):
raise DiscussionValidationError("message.member text must be a non-pass string")
member = _member_by_id(room, payload.get("member_id"))
peer = _peer_target(member)
expected = {
"kind": "member", "id": member.member_id, "profile": member.profile,
"connection_id": peer.get("peer_id") if peer else None}
expected = {**_member_actor(member, display_name=False), "connection_id": _peer_id(member)}
if any(actor.get(key) != value for key, value in expected.items()):
raise DiscussionValidationError("message.member actor does not match roster")
return payload
def _validate_terminal_event(
kind: str, payload: Mapping[str, Any], actor: Mapping[str, Any], room: DiscussionRoom) -> Mapping[str, Any]:
def _validate_terminal_event(kind: str, payload: Payload, actor: Payload, room: DiscussionRoom) -> Payload:
_exact_fields(
payload, label=f"{kind} payload", required=_TERMINAL_COMMON_FIELDS | _TERMINAL_EXTRA_FIELDS[kind],
payload, label=f"{kind} payload", required=_TERMINAL_FIELDS[kind],
optional=_TERMINAL_OPTIONAL_FIELDS.get(kind, frozenset()))
_validate_turn_coordinates(payload, room)
_positive_int(payload.get("seen_through_seq"), label="seen_through_seq")
@@ -404,15 +404,12 @@ def _validate_terminal_event(
raise DiscussionValidationError(f"{kind} {field} must be non-empty")
if kind == "turn.deferred":
_positive_int(payload.get("execution_generation"), label="execution_generation")
if kind == "turn.failed" and "reason_code" in payload:
from tools.bot_failure_reasons import ALL_REASONS
if payload["reason_code"] not in ALL_REASONS:
raise DiscussionValidationError("turn.failed reason_code must use the shared failure vocabulary")
if kind == "turn.failed" and "reason_code" in payload and payload["reason_code"] not in _all_failure_reasons():
raise DiscussionValidationError("turn.failed reason_code must use the shared failure vocabulary")
return payload
def _validate_gateway_event(
kind: str, payload: Mapping[str, Any], actor: Mapping[str, Any], room: DiscussionRoom) -> Mapping[str, Any]:
def _validate_gateway_event(kind: str, payload: Payload, actor: Payload, room: DiscussionRoom) -> Payload:
fields, identifier_fields = _GATEWAY_EVENT_FIELDS[kind]
_exact_fields(payload, label=f"{kind} payload", required=fields)
if kind == "room.activity" and payload.get("status") not in {"settled", "bounded"}:
@@ -434,8 +431,7 @@ def _validate_event(raw: Any, *, room: DiscussionRoom, previous_seq: int) -> _Va
raise DiscussionValidationError("room event must be an object")
if raw.get("room_id") != room.room_id:
raise DiscussionValidationError("room event belongs to a different room")
seq = _positive_int(raw.get("seq"), label="event seq")
if seq <= previous_seq:
if (seq := _positive_int(raw.get("seq"), label="event seq")) <= previous_seq:
raise DiscussionValidationError("room events must be in strict sequence order")
event_id = _identifier(raw.get("event_id"), label="event_id")
kind, actor, payload = raw.get("kind"), raw.get("actor"), raw.get("payload")
@@ -446,22 +442,19 @@ def _validate_event(raw: Any, *, room: DiscussionRoom, previous_seq: int) -> _Va
raise DiscussionValidationError(message)
if kind in _EPOCH_STAMPED_KINDS and raw.get("authority_epoch") != room.authority_epoch:
raise DiscussionValidationError(f"{kind} authority epoch does not match the room")
validator = _EVENT_VALIDATORS.get(kind)
if validator is not None:
if (validator := _EVENT_VALIDATORS.get(kind)) is not None:
payload = validator(kind, payload, actor, room)
return _ValidatedEvent(raw=raw, seq=seq, event_id=event_id, kind=kind, actor=actor, payload=payload)
def _validated_events(events: Sequence[Mapping[str, Any]], *, room: DiscussionRoom) -> tuple[_ValidatedEvent, ...]:
validated: list[_ValidatedEvent] = []
previous_seq = 0
event_ids: set[str] = set()
for raw in events:
event = _validate_event(raw, room=room, previous_seq=previous_seq)
event = _validate_event(raw, room=room, previous_seq=validated[-1].seq if validated else 0)
if event.event_id in event_ids:
raise DiscussionValidationError("room event ids must be unique")
validated.append(event)
previous_seq = event.seq
event_ids.add(event.event_id)
return tuple(validated)
@@ -470,8 +463,8 @@ def derive_member_watermarks(
room_value: Any, events: Sequence[Mapping[str, Any]], *, local_profiles: Iterable[str]
) -> dict[tuple[str, str], int]:
"""Derive ``(thread_id, member_id)`` watermarks from terminal events."""
room = validate_room(room_value, local_profiles=local_profiles)
return _derive_member_watermarks(_validated_events(events, room=room))
return _derive_member_watermarks(
_validated_events(events, room=validate_room(room_value, local_profiles=local_profiles)))
def _derive_member_watermarks(events: Sequence[_ValidatedEvent]) -> dict[tuple[str, str], int]:
@@ -505,38 +498,27 @@ def _derive_member_watermarks(events: Sequence[_ValidatedEvent]) -> dict[tuple[s
def _member_digest(member: DiscussionMember) -> str:
target = compact_json(member.target or {"kind": "local", "profile": member.profile}, ensure_ascii=False)
seed = f"{member.member_id}\0{member.profile}\0{member.handle}\0{target}"
return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24]
return hashlib.sha256(f"{member.member_id}\0{member.profile}\0{member.handle}\0{target}".encode()).hexdigest()[:24]
def _rotate(members: Sequence[DiscussionMember], round_index: int) -> tuple[DiscussionMember, ...]:
if len(members) < 2:
return tuple(members)
shift = round_index % len(members)
shift = round_index % len(members) if members else 0
return tuple((*members[shift:], *members[:shift]))
def _format_message(event: _ValidatedEvent, room: DiscussionRoom) -> str:
text = str(event.payload["text"])
if event.kind == "message.user":
return f"User (user): {text}"
member = _member_by_id(room, event.payload["member_id"])
return f"@{member.handle}: {text}"
return f"User (user): {event.payload['text']}"
return f"@{_member_by_id(room, event.payload['member_id']).handle}: {event.payload['text']}"
def _truncate_utf8_text(value: Any, *, max_bytes: int, suffix: str = "") -> str:
text = str(value or "")
encoded = text.encode("utf-8")
if len(encoded) <= max_bytes:
if len(encoded := text.encode("utf-8")) <= max_bytes:
return text
suffix_bytes = suffix.encode("utf-8")
prefix = encoded[: max(0, max_bytes - len(suffix_bytes))]
while prefix:
try:
return prefix.decode("utf-8") + suffix
except UnicodeDecodeError:
prefix = prefix[:-1]
return suffix.strip()
# ``ignore`` drops only the incomplete trailing sequence of a cut valid-UTF-8 prefix.
prefix = encoded[: max(0, max_bytes - len(suffix.encode("utf-8")))].decode("utf-8", "ignore")
return prefix + suffix if prefix else suffix.strip()
def _build_prompt(
@@ -560,8 +542,7 @@ def _build_prompt(
omitted = False
for event in reversed(delta):
line = f" {_format_message(event, room)}"
line_bytes = len(line.encode("utf-8")) + 1
if line_bytes <= available:
if (line_bytes := len(line.encode("utf-8")) + 1) <= available:
selected.append(line)
available -= line_bytes
continue
@@ -572,8 +553,7 @@ def _build_prompt(
selected.reverse()
if omitted:
selected.insert(0, " [Earlier content omitted to fit this turn.]")
prompt = "\n".join([*opening, *selected, *rules])
if len(prompt.encode("utf-8")) > driver.MAX_PROMPT_BYTES:
if len((prompt := "\n".join([*opening, *selected, *rules])).encode("utf-8")) > driver.MAX_PROMPT_BYTES:
raise DiscussionValidationError("Discussion prompt exceeds the driver limit")
return prompt
@@ -581,23 +561,20 @@ def _build_prompt(
def _make_task_plan(
*, room: DiscussionRoom, discussion_event: _ValidatedEvent, member: DiscussionMember, member_index: int,
round_index: int, seen_through_seq: int, prompt: str) -> DiscussionTaskPlan:
turn_id = (
f"d{discussion_event.seq}.r{round_index}.p{member_index}."
f"s{seen_through_seq}.m{_member_digest(member)}")
turn_id = f"d{discussion_event.seq}.r{round_index}.p{member_index}.s{seen_through_seq}.m{_member_digest(member)}"
seed = compact_json({
"discussion_event_id": discussion_event.event_id, "member_id": member.member_id, "member_index": member_index,
"prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(), "room_id": room.room_id,
"round_index": round_index, "seen_through_seq": seen_through_seq, "source_event_seq": discussion_event.seq,
"thread_id": discussion_event.payload["thread_id"]})
task_id = f"dtask:{hashlib.sha256(seed.encode('utf-8')).hexdigest()[:48]}"
identity = driver.TaskIdentity(
room_id=room.room_id, task_id=task_id, thread_id=str(discussion_event.payload["thread_id"]), turn_id=turn_id)
room_id=room.room_id, task_id=f"dtask:{hashlib.sha256(seed.encode('utf-8')).hexdigest()[:48]}",
thread_id=str(discussion_event.payload["thread_id"]), turn_id=turn_id)
payload = {
"target_member_id": member.member_id, "target_profile": member.profile, "prompt": prompt,
"source_event_seq": discussion_event.seq}
return DiscussionTaskPlan(
identity=identity, payload=payload, discussion_event_id=discussion_event.event_id, member=member,
member_index=member_index, round_index=round_index, seen_through_seq=seen_through_seq)
identity, payload, discussion_event.event_id, member, member_index, round_index, seen_through_seq)
def _pending_discussion(validated: Sequence[_ValidatedEvent]) -> _ValidatedEvent | None:
@@ -607,15 +584,11 @@ def _pending_discussion(validated: Sequence[_ValidatedEvent]) -> _ValidatedEvent
str(event.payload["discussion_event_id"])
for event in validated
if event.kind == "room.activity" and event.payload.get("status") in {"settled", "bounded"}}
latest_by_thread: dict[str, _ValidatedEvent] = {}
for event in validated:
if event.kind == "message.user":
latest_by_thread[str(event.payload["thread_id"])] = event
pending = [
event
for event in sorted(latest_by_thread.values(), key=lambda item: item.seq)
if event.seq > stopped_through_seq and event.event_id not in completed_discussion_ids]
return pending[0] if pending else None
latest_by_thread = {
str(event.payload["thread_id"]): event for event in validated if event.kind == "message.user"}
return next((
event for event in sorted(latest_by_thread.values(), key=lambda item: item.seq)
if event.seq > stopped_through_seq and event.event_id not in completed_discussion_ids), None)
def _thread_messages(
@@ -651,9 +624,7 @@ def _effective_watermarks(
) -> dict[tuple[str, str], int]:
watermarks = {
(str(thread_id), str(member_id)): int(value)
for (thread_id, member_id),
value in (initial_watermarks or {}).items()
if int(value) >= 0}
for (thread_id, member_id), value in (initial_watermarks or {}).items() if int(value) >= 0}
for key, value in _derive_member_watermarks(validated).items():
watermarks[key] = max(watermarks.get(key, 0), value)
return watermarks
@@ -718,15 +689,14 @@ def reconstruct_task_plan(
"""Reconstruct and verify one persisted driver task after a restart."""
room = validate_room(room_value, local_profiles=local_profiles)
validated = _validated_events(events, room=room)
identity = task.get("identity")
payload = task.get("payload")
identity, payload = task.get("identity"), task.get("payload")
if not isinstance(identity, driver.TaskIdentity) or not isinstance(payload, Mapping):
raise DiscussionReconstructionError("driver task has no valid identity or payload")
required_payload = frozenset({"target_profile", "prompt", "source_event_seq"})
if not required_payload <= frozenset(payload) or (frozenset(payload) - required_payload - {"target_member_id"}):
fields = frozenset(payload)
if not driver._TASK_PAYLOAD_REQUIRED_FIELDS <= fields or (
fields - driver._TASK_PAYLOAD_REQUIRED_FIELDS - driver._TASK_PAYLOAD_OPTIONAL_FIELDS):
raise DiscussionReconstructionError("driver task payload shape changed")
match = _TURN_ID_RE.fullmatch(identity.turn_id)
if match is None:
if (match := _TURN_ID_RE.fullmatch(identity.turn_id)) is None:
raise DiscussionReconstructionError("turn_id is not a Discussion coordinate")
source_event_seq = int(match.group("source"))
if payload.get("source_event_seq") != source_event_seq:
@@ -736,13 +706,11 @@ def reconstruct_task_plan(
raise DiscussionReconstructionError("task source user event is missing")
if identity.room_id != room.room_id or identity.thread_id != discussion.payload["thread_id"]:
raise DiscussionReconstructionError("task identity does not match its room thread")
profile = payload.get("target_profile")
target_member_id = payload.get("target_member_id")
member = next(
(
candidate for candidate in room.members
if candidate.profile == profile
and (target_member_id is None or candidate.member_id == target_member_id)), None)
profile, target_member_id = payload.get("target_profile"), payload.get("target_member_id")
member = next((
candidate for candidate in room.members
if candidate.profile == profile and (target_member_id is None or candidate.member_id == target_member_id)
), None)
if member is None or _member_digest(member) != match.group("member"):
raise DiscussionReconstructionError("task target member does not match turn_id")
prompt = payload.get("prompt")
@@ -768,13 +736,8 @@ def _turn_coordinates(task: DiscussionTaskPlan) -> dict[str, Any]:
def _terminal_text(result: Any, *, field: str, fallback: str) -> str:
if isinstance(result, Mapping):
value = result.get(field)
if value is None and field == "error":
value = result.get("text")
else:
value = result
text = str(value or "").strip()
return text or fallback
result = result.get(field) if field != "error" or result.get("error") is not None else result.get("text")
return str(result or "").strip() or fallback
# -- per-status terminal payload builders (dispatched by plan_publication) -----
@@ -787,37 +750,27 @@ def _settled_effects(
text = _truncate_utf8_text(
_terminal_text(result, field="text", fallback=""), max_bytes=MAX_MEMBER_TEXT_BYTES,
suffix=_TRUNCATED_REPLY_NOTICE)
passed = is_pass_text(text)
effects: list[EventPlan] = []
if not passed:
member_actor = {"kind": "member", "id": task.member.member_id, "profile": task.member.profile}
peer = _peer_target(task.member)
if peer:
member_actor["connection_id"] = peer["peer_id"]
if task.member.display_name:
member_actor["display_name"] = task.member.display_name
effects.append(EventPlan(
event_id=message_event_id, kind="message.member", actor=member_actor,
payload={**_turn_coordinates(task), "text": text}, authority_gateway_id=room.gateway_id,
authority_epoch=room.authority_epoch))
return {"message_event_id": None if passed else message_event_id, "passed": passed}, effects
if is_pass_text(text):
return {"message_event_id": None, "passed": True}, []
return {"message_event_id": message_event_id, "passed": False}, [EventPlan(
event_id=message_event_id, kind="message.member", actor=_member_actor(task.member),
payload={**_turn_coordinates(task), "text": text}, authority_gateway_id=room.gateway_id,
authority_epoch=room.authority_epoch)]
def _failed_effects(result: Any, **_: Any) -> tuple[dict[str, Any], list[EventPlan]]:
error_text = _terminal_text(result, field="error", fallback="member turn failed")
from tools.bot_failure_reasons import ALL_REASONS, classify_agent_error
from tools.bot_failure_reasons import classify_agent_error
supplied_reason = (
str(result.get("reason_code") or result.get("reason") or "").strip()
if isinstance(result, Mapping) else "")
reason_code = supplied_reason if supplied_reason in ALL_REASONS else classify_agent_error(error_text)
str(result.get("reason_code") or result.get("reason") or "").strip() if isinstance(result, Mapping) else "")
reason_code = supplied_reason if supplied_reason in _all_failure_reasons() else classify_agent_error(error_text)
return {"error": error_text, "reason_code": reason_code}, []
def _cancelled_effects(result: Any, *, newer_same_thread: bool, ** _: Any) -> tuple[dict[str, Any], list[EventPlan]]:
reason = (
"superseded_by_newer_user_event" if newer_same_thread
else _terminal_text(result, field="reason", fallback="member turn cancelled"))
return {"reason": reason}, []
def _cancelled_effects(result: Any, *, newer_same_thread: bool, **_: Any) -> tuple[dict[str, Any], list[EventPlan]]:
if newer_same_thread:
return {"reason": "superseded_by_newer_user_event"}, []
return {"reason": _terminal_text(result, field="reason", fallback="member turn cancelled")}, []
def _deferred_effects(
@@ -850,8 +803,7 @@ def plan_publication(
if status not in _TERMINAL_EFFECTS:
raise DiscussionValidationError("invalid terminal publication status")
if status == "deferred":
message = "deferred publication requires an execution generation"
_bounded_int(execution_generation, message=message, low=1)
_bounded_int(execution_generation, message="deferred publication requires an execution generation", low=1)
newer_same_thread = any(
event.kind == "message.user"
and event.seq > task.seen_through_seq
@@ -860,15 +812,13 @@ def plan_publication(
effective_status: TerminalKind = ("cancelled" if newer_same_thread and status != "deferred" else status)
digest = task.identity.task_id.removeprefix("dtask:")
terminal_event_id = (
f"ddeferred:{digest}:g{execution_generation}"
if effective_status == "deferred" else f"dterminal:{digest}")
common_payload = {**_turn_coordinates(task), "seen_through_seq": task.seen_through_seq}
f"ddeferred:{digest}:g{execution_generation}" if effective_status == "deferred" else f"dterminal:{digest}")
extra, effects = _TERMINAL_EFFECTS[effective_status](
result, task=task, room=room, message_event_id=f"dmessage:{digest}", newer_same_thread=newer_same_thread,
execution_generation=execution_generation)
terminal_kind = f"turn.{effective_status}"
effects.append(EventPlan(
event_id=terminal_event_id, kind=terminal_kind, actor={"kind": "gateway", "id": room.gateway_id},
payload={**common_payload, **extra}, authority_gateway_id=room.gateway_id, authority_epoch=room.authority_epoch
))
payload={**_turn_coordinates(task), "seen_through_seq": task.seen_through_seq, **extra},
authority_gateway_id=room.gateway_id, authority_epoch=room.authority_epoch))
return PublicationPlan(task_id=task.identity.task_id, terminal_kind=terminal_kind, events=tuple(effects))
gateway/hosted_room_peer.py
+50 -67
View File
@@ -50,9 +50,7 @@ _ROOM_GRANT_SECRET_FILE = ".room-link-grant-secret"
def _gateway_room_grant_secret_for_home(home_value: str) -> bytes:
"""Load one restart-scoped grant secret for an exact installation root."""
from hermes_cli.install_identity import _fsync_directory
home = Path(home_value)
home.mkdir(parents=True, exist_ok=True)
(home := Path(home_value)).mkdir(parents=True, exist_ok=True)
path = home / _ROOM_GRANT_SECRET_FILE
def _read() -> bytes:
data = path.read_bytes()
@@ -94,7 +92,6 @@ def gateway_room_grant_secret(root: Path | str | None = None) -> bytes:
"""
if root is None:
from hermes_constants import get_hermes_home
# Profile routing uses a context-local HERMES_HOME override; the process environment
# retains the installation root and is the authority here.
root = os.environ.get("HERMES_HOME") or get_hermes_home()
@@ -147,21 +144,35 @@ def _b64decode(value: str) -> bytes:
raise HostedRoomGrantError("room grant encoding is invalid") from exc
def _split_token(token: str) -> tuple[str, str]:
def _split_token(token: str) -> tuple[bytes, bytes]:
"""Return the decoded ``(payload, signature)`` halves of a grant token."""
encoded_token, separator, signature_token = token.partition(".")
if not separator:
raise HostedRoomGrantError("room grant is invalid")
return encoded_token, signature_token
return _b64decode(encoded_token), _b64decode(signature_token)
def _non_empty_list(value: Any, *, field: str) -> list[Any]:
if not isinstance(value, list) or not value:
raise HostedRoomPeerError(f"{field} must be a non-empty list")
return value
def _parse_link_modes(links_raw: Any) -> tuple[LinkMode, ...]:
if not isinstance(links_raw, list) or not links_raw:
raise HostedRoomPeerError("link_modes must be a non-empty list")
if any(item not in _LINK_MODES for item in links_raw):
if any(item not in _LINK_MODES for item in _non_empty_list(links_raw, field="link_modes")):
raise HostedRoomPeerError("link_modes contains an unsupported mode")
return tuple(dict.fromkeys(links_raw))
def _protocol_versions(versions_raw: Any) -> list[int]:
return sorted({_positive_int(item, field="protocol_version") for item in versions_raw})
def _catalog_digest(value: Mapping[str, Any]) -> str:
"""Digest of the catalog mapping without its ``catalog_digest`` field."""
return hashlib.sha256(_canonical_json({k: v for k, v in value.items() if k != "catalog_digest"})).hexdigest()
def _parse_endpoint(endpoint: Any) -> tuple[str | None, str | None, TransportSecurity | None]:
"""Return ``(url, reason, transport_security)`` for an advertised endpoint capability."""
if not isinstance(endpoint, Mapping) or not isinstance(endpoint.get("available"), bool):
@@ -201,27 +212,17 @@ class GatewayRoomCatalog:
def from_mapping(cls, value: Mapping[str, Any]) -> "GatewayRoomCatalog":
_exact_fields(value, required=_CATALOG_FIELDS, optional={"endpoint"}, label="capability catalog")
installation_id = _identifier(value["installation_id"], field="installation_id")
versions_raw = value["protocol_versions"]
if not isinstance(versions_raw, list) or not versions_raw:
raise HostedRoomPeerError("protocol_versions must be a non-empty list")
versions = tuple(sorted({_positive_int(item, field="protocol_version") for item in versions_raw}))
versions = tuple(_protocol_versions(_non_empty_list(value["protocol_versions"], field="protocol_versions")))
links = _parse_link_modes(value["link_modes"])
for field in ("persistent_process", "text", "attachments"):
if not isinstance(value[field], bool):
raise HostedRoomPeerError(f"{field} must be a boolean")
policy = RoomExecutionPolicy.from_mapping(value["execution_policy"])
endpoint_url = endpoint_reason = transport_security = None
if "endpoint" in value:
endpoint_url, endpoint_reason, transport_security = _parse_endpoint(value["endpoint"])
endpoint = _parse_endpoint(value["endpoint"]) if "endpoint" in value else (None, None, None)
catalog = cls(
installation_id=installation_id, protocol_versions=versions, link_modes=links,
persistent_process=value["persistent_process"], text=value["text"], attachments=value["attachments"],
execution_policy=policy, catalog_digest=_digest(value["catalog_digest"], field="catalog_digest"),
endpoint_url=endpoint_url, endpoint_reason=endpoint_reason, transport_security=transport_security)
unsigned = catalog.as_mapping()
del unsigned["catalog_digest"]
expected = hashlib.sha256(_canonical_json(unsigned)).hexdigest()
if not hmac.compare_digest(expected, catalog.catalog_digest):
installation_id, versions, links, value["persistent_process"], value["text"], value["attachments"],
policy, _digest(value["catalog_digest"], field="catalog_digest"), *endpoint)
if not hmac.compare_digest(_catalog_digest(catalog.as_mapping()), catalog.catalog_digest):
raise HostedRoomPeerError("catalog_digest does not match the catalog")
return catalog
@@ -238,9 +239,9 @@ class GatewayRoomCatalog:
def endpoint_mapping(self) -> dict[str, Any]:
"""Return the normalized self-advertised endpoint capability."""
if self.endpoint_url is not None:
return {"available": True, "url": self.endpoint_url, "transport_security": self.transport_security}
return {"available": False, "reason": self.endpoint_reason or "not_configured"}
if self.endpoint_url is None:
return {"available": False, "reason": self.endpoint_reason or "not_configured"}
return {"available": True, "url": self.endpoint_url, "transport_security": self.transport_security}
def catalog_mapping(
@@ -260,13 +261,13 @@ def catalog_mapping(
raise HostedRoomPeerError("remote room execution requires manual or smart approvals")
value = {
"installation_id": _identifier(installation_id, field="installation_id"),
"protocol_versions": sorted({_positive_int(item, field="protocol_version") for item in protocol_versions}),
"protocol_versions": _protocol_versions(protocol_versions),
# Direct HTTPS/loopback is the only implemented RoomLink transport; never advertise pull/relay.
"link_modes": [mode for mode in dict.fromkeys(link_modes) if mode == "direct"],
"persistent_process": persistent_process, "text": bool(text), "attachments": bool(attachments),
"execution_policy": checked_policy.as_mapping(),
"endpoint": dict(local_room_link_endpoint() if endpoint is None else endpoint)}
value["catalog_digest"] = hashlib.sha256(_canonical_json(value)).hexdigest()
value["catalog_digest"] = _catalog_digest(value)
GatewayRoomCatalog.from_mapping(value)
return value
@@ -277,8 +278,7 @@ local_catalog_mapping = partial(catalog_mapping, persistent_process=True)
def local_room_link_endpoint(value: Any | None = None) -> dict[str, Any]:
"""Return the validated endpoint this gateway explicitly advertises."""
configured = _configured_room_link_url() if value is None else value
if not str(configured or "").strip():
if not str((configured := _configured_room_link_url() if value is None else value) or "").strip():
return {"available": False, "reason": "not_configured"}
try:
url, transport_security = validate_room_link_url(configured)
@@ -305,13 +305,11 @@ def _room_link_url_from_config(home: str) -> str | None:
def _configured_room_link_url() -> str | None:
"""Resolve the explicit endpoint: env override > profile config > root config."""
override = os.getenv("HERMES_ROOM_LINK_URL")
if override is not None:
if (override := os.getenv("HERMES_ROOM_LINK_URL")) is not None:
return override
from hermes_constants import get_default_hermes_root, get_hermes_home
home = get_hermes_home()
configured = _room_link_url_from_config(str(home))
if configured:
if configured := _room_link_url_from_config(str(home)):
return configured
# RoomLink is a gateway reachability property, not a Bot personality setting: named profiles may
# override it but otherwise inherit the process gateway's root endpoint.
@@ -337,15 +335,12 @@ def validate_room_link_url(value: Any) -> tuple[str, TransportSecurity]:
return raw, "tls"
if parsed.scheme.lower() != "http":
raise HostedRoomPeerError("target_url must use https")
loopback = hostname == "localhost" or hostname.endswith(".localhost")
if not loopback:
try:
loopback = ipaddress.ip_address(hostname).is_loopback
except ValueError:
loopback = False
if not loopback:
raise HostedRoomPeerError("target_url must use https outside the local machine")
return raw, "loopback"
try:
if hostname == "localhost" or hostname.endswith(".localhost") or ipaddress.ip_address(hostname).is_loopback:
return raw, "loopback"
except ValueError:
pass
raise HostedRoomPeerError("target_url must use https outside the local machine")
# Validator per dispatch field, in validation order (prompt/prompt_digest are cross-checked in from_mapping).
@@ -389,9 +384,8 @@ class HostedMemberDispatch:
if not isinstance(prompt, str) or not prompt.strip():
raise HostedRoomPeerError("prompt must be a non-empty string")
text(prompt, error=HostedRoomPeerError, label="prompt", max_bytes=MAX_PROMPT_BYTES, strip=False)
expected_prompt_digest = hashlib.sha256(prompt.encode("utf-8")).hexdigest()
prompt_digest = _digest(value["prompt_digest"], field="prompt_digest")
if not hmac.compare_digest(expected_prompt_digest, prompt_digest):
if not hmac.compare_digest(hashlib.sha256(prompt.encode("utf-8")).hexdigest(), prompt_digest):
raise HostedRoomPeerError("prompt_digest does not match prompt")
return cls(
prompt=prompt, prompt_digest=prompt_digest,
@@ -432,10 +426,7 @@ def issue_room_grant(
allowed = tuple(sorted(set(permissions)))
if not allowed or not set(allowed) <= _GRANT_PERMISSIONS:
raise HostedRoomGrantError("room grant permissions are invalid")
scope = {
"grant_id": grant_id, "room_id": room_id, "home_install_id": home_install_id,
"authority_gateway_id": authority_gateway_id, "authority_epoch": authority_epoch, "member_id": member_id,
"target_install_id": target_install_id, "target_profile": target_profile}
scope = locals()
payload = {
"version": PROTOCOL_VERSION,
**{name: _DISPATCH_FIELDS.get(name, _identifier)(scope[name], field=name) for name in _GRANT_SCOPE},
@@ -445,8 +436,7 @@ def issue_room_grant(
), "permissions": list(allowed), "issued_at": now, "expires_at": now + float(ttl_seconds),
"status_expires_at": bounded_status_expiry}
encoded = _canonical_json(payload)
signature = hmac.new(secret, encoded, hashlib.sha256).digest()
token = f"{_b64encode(encoded)}.{_b64encode(signature)}"
token = f"{_b64encode(encoded)}.{_b64encode(hmac.new(secret, encoded, hashlib.sha256).digest())}"
if len(token.encode("ascii")) > MAX_TOKEN_BYTES:
raise HostedRoomGrantError("room grant is too large")
return token
@@ -459,8 +449,7 @@ def verify_room_grant(
payload = decode_room_grant(secret, token, permission=permission, now=now)
if payload["version"] != dispatch.protocol_version:
raise HostedRoomGrantError("room grant protocol does not match dispatch")
scope = (*_GRANT_SCOPE[1:], "execution_policy_digest")
if any(payload.get(field) != getattr(dispatch, field) for field in scope):
if any(payload.get(f) != getattr(dispatch, f) for f in (*_GRANT_SCOPE[1:], "execution_policy_digest")):
raise HostedRoomGrantError("room grant scope does not match dispatch")
return payload
@@ -469,11 +458,8 @@ def decode_room_grant(secret: bytes, token: str, *, permission: str, now: float
"""Verify grant signature, lifetime and operation without a dispatch."""
if not isinstance(token, str) or len(token.encode("utf-8")) > MAX_TOKEN_BYTES:
raise HostedRoomGrantError("room grant is invalid")
encoded_token, signature_token = _split_token(token)
encoded = _b64decode(encoded_token)
supplied_signature = _b64decode(signature_token)
expected_signature = hmac.new(secret, encoded, hashlib.sha256).digest()
if not hmac.compare_digest(expected_signature, supplied_signature):
encoded, supplied_signature = _split_token(token)
if not hmac.compare_digest(hmac.new(secret, encoded, hashlib.sha256).digest(), supplied_signature):
raise HostedRoomGrantError("room grant signature is invalid")
try:
payload = json.loads(encoded.decode("ascii"))
@@ -481,8 +467,7 @@ def decode_room_grant(secret: bytes, token: str, *, permission: str, now: float
raise HostedRoomGrantError("room grant payload is invalid") from exc
if not isinstance(payload, dict) or frozenset(payload) not in {_GRANT_FIELDS, _GRANT_REFRESH_FIELDS}:
raise HostedRoomGrantError("room grant fields are invalid")
checked_now = clock(now)
if not math.isfinite(checked_now):
if not math.isfinite(checked_now := clock(now)):
raise HostedRoomGrantError("room grant clock is invalid")
try:
issued_at = float(payload["issued_at"])
@@ -490,8 +475,8 @@ def decode_room_grant(secret: bytes, token: str, *, permission: str, now: float
status_expires_at = float(payload.get("status_expires_at", expires_at))
except (TypeError, ValueError) as exc:
raise HostedRoomGrantError("room grant lifetime is invalid") from exc
finite = all(map(math.isfinite, (issued_at, expires_at, status_expires_at)))
if not (finite and issued_at < expires_at <= status_expires_at):
lifetimes = (issued_at, expires_at, status_expires_at)
if not (all(map(math.isfinite, lifetimes)) and issued_at < expires_at <= status_expires_at):
raise HostedRoomGrantError("room grant lifetime is invalid")
operation_expires_at = status_expires_at if permission in {"approve", "status", "stop"} else expires_at
if checked_now < issued_at - 30 or checked_now >= operation_expires_at:
@@ -505,9 +490,7 @@ def decode_room_grant(secret: bytes, token: str, *, permission: str, now: float
def room_grant_needs_dispatch_refresh(token: str, *, now: float | None = None, leeway_seconds: float = 5 * 60) -> bool:
"""Read only grant timing to schedule refresh; trust is established by the target, not here."""
try:
payload = json.loads(_b64decode(_split_token(token)[0]).decode("ascii"))
expires_at = float(payload["expires_at"])
checked_now = clock(now)
return checked_now + max(0.0, float(leeway_seconds)) >= expires_at
expires_at = float(json.loads(_split_token(token)[0].decode("ascii"))["expires_at"])
return clock(now) + max(0.0, float(leeway_seconds)) >= expires_at
except Exception:
return True