diff --git a/gateway/hosted_room_discussion.py b/gateway/hosted_room_discussion.py index 93e2c4d961..af376853a9 100644 --- a/gateway/hosted_room_discussion.py +++ b/gateway/hosted_room_discussion.py @@ -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[0-5])\.s(?P[1-9][0-9]*)\." r"m(?P[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)) diff --git a/gateway/hosted_room_peer.py b/gateway/hosted_room_peer.py index 4bf72550a1..21a005d637 100644 --- a/gateway/hosted_room_peer.py +++ b/gateway/hosted_room_peer.py @@ -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