diff --git a/gateway/hosted_room_discussion.py b/gateway/hosted_room_discussion.py index 6e399bae30..b4aac19951 100644 --- a/gateway/hosted_room_discussion.py +++ b/gateway/hosted_room_discussion.py @@ -18,10 +18,12 @@ import json import re from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass +from functools import partial from typing import Any, Literal from gateway import hosted_room_driver as driver from gateway import hosted_rooms +from gateway import hosted_rooms_common as common MAX_DISCUSSION_MEMBERS = 6 @@ -31,14 +33,11 @@ MAX_DISCUSSION_MESSAGES = 10 MAX_DISCUSSION_DELTA_LINES = 24 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.]" -) +_TRUNCATED_REPLY_NOTICE = "\n\n[Reply truncated. Ask the Bot to share the full result as a file.]" DecisionStatus = Literal["idle", "task", "settled", "bounded"] TerminalKind = Literal["settled", "failed", "cancelled", "deferred"] -_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") _MENTION_RE = re.compile(r"@([A-Za-z0-9][A-Za-z0-9._:-]*)", re.IGNORECASE) _TURN_ID_RE = re.compile( r"^d(?P[1-9][0-9]*)\.r(?P[0-2])\." @@ -47,9 +46,7 @@ _TURN_ID_RE = re.compile( ) _LOCAL_TARGET_FIELDS = frozenset({"kind", "profile"}) -_PEER_TARGET_FIELDS = frozenset( - {"kind", "peer_id", "installation_id", "profile", "capability_digest"} -) +_PEER_TARGET_FIELDS = frozenset({"kind", "peer_id", "installation_id", "profile", "capability_digest"}) _REMOTE_MEMBER_FIELDS = frozenset({ "connectionId", "connectionKind", "connectionLabel", "connection_id", "connection_kind", "connection_label", "remoteSource", "route", @@ -155,11 +152,8 @@ class EventPlan: 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), + "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, } @@ -184,25 +178,16 @@ class _ValidatedEvent: payload: Mapping[str, Any] -def _identifier(value: Any, *, label: str) -> str: - if not isinstance(value, str): - raise DiscussionValidationError(f"{label} must be a string") - normalized = value.strip() - if ( - not normalized - or len(normalized) > driver.MAX_IDENTIFIER_CHARS - or not _IDENTIFIER_RE.fullmatch(normalized) - ): - raise DiscussionValidationError(f"invalid {label}") - return normalized +_identifier = partial( + common.identifier, error=DiscussionValidationError, max_chars=driver.MAX_IDENTIFIER_CHARS +) +_exact_fields = partial(common.exact_fields, error=DiscussionValidationError) -def _positive_int(value: Any, *, label: str, maximum: int | None = None) -> int: - if isinstance(value, bool) or not isinstance(value, int) or value < 1: - raise DiscussionValidationError(f"{label} must be a positive integer") - if maximum is not None and value > maximum: - raise DiscussionValidationError(f"{label} must be at most {maximum}") - return value +def _positive_int(value: Any, *, label: str) -> int: + return common.positive_int( + value, error=DiscussionValidationError, message=f"{label} must be a positive integer" + ) def _zero_based_int(value: Any, *, label: str, maximum: int) -> int: @@ -211,25 +196,6 @@ def _zero_based_int(value: Any, *, label: str, maximum: int) -> int: return value -def _exact_fields( - value: Any, - *, - label: str, - required: frozenset[str], - optional: frozenset[str] = frozenset(), -) -> Mapping[str, Any]: - if not isinstance(value, Mapping): - raise DiscussionValidationError(f"{label} must be an object") - keys = frozenset(value) - missing = required - keys - unknown = keys - required - optional - if missing: - raise DiscussionValidationError(f"{label} is missing fields: {', '.join(sorted(missing))}") - if unknown: - raise DiscussionValidationError(f"{label} has unknown fields: {', '.join(sorted(unknown))}") - return value - - def _peer_target(member: DiscussionMember) -> Mapping[str, Any] | None: target = member.target return target if target and target.get("kind") == "peer" else None @@ -264,8 +230,7 @@ def _validate_member_target( if kind not in ("local", "peer"): raise DiscussionValidationError(f"member {index} target kind must be local or peer") target = _exact_fields( - value, - label=f"member {index} {kind} target", + value, label=f"member {index} {kind} target", required=_LOCAL_TARGET_FIELDS if kind == "local" else _PEER_TARGET_FIELDS, ) target_profile = _identifier(target["profile"], label=f"member {index} target profile") @@ -280,26 +245,18 @@ def _validate_member_target( f"member {index} peer target profile does not match member profile" ) capability_digest = target["capability_digest"] - if not isinstance(capability_digest, str) or not re.fullmatch( - r"[0-9a-f]{64}", capability_digest - ): - raise DiscussionValidationError( - f"member {index} capability_digest must be a sha256 digest" - ) + if not isinstance(capability_digest, str) or not re.fullmatch(r"[0-9a-f]{64}", capability_digest): + raise DiscussionValidationError(f"member {index} capability_digest must be a sha256 digest") return { "kind": "peer", "peer_id": _identifier(target["peer_id"], label=f"member {index} peer_id"), - "installation_id": _identifier( - target["installation_id"], label=f"member {index} installation_id" - ), + "installation_id": _identifier(target["installation_id"], label=f"member {index} installation_id"), "profile": target_profile, "capability_digest": capability_digest, } -def validate_roster( - value: Any, *, local_profiles: Iterable[str] -) -> tuple[DiscussionMember, ...]: +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") @@ -312,7 +269,7 @@ def validate_roster( known_profiles = {_identifier(profile, label="local profile") for profile in local_profiles} members: list[DiscussionMember] = [] targets: set[str] = set() - handles: set[str] = set() + handles: set[str] = {"all", "everyone"} # reserved mention handles member_ids: set[str] = set() for index, raw in enumerate(value): @@ -321,12 +278,10 @@ def validate_roster( remote_fields = frozenset(raw) & _REMOTE_MEMBER_FIELDS if remote_fields: raise DiscussionValidationError( - f"member {index} contains cross-gateway fields: " - f"{', '.join(sorted(remote_fields))}" + f"member {index} contains cross-gateway fields: {', '.join(sorted(remote_fields))}" ) member = _exact_fields( - raw, - label=f"member {index}", + raw, label=f"member {index}", required=frozenset({"member_id", "profile", "handle"}), optional=frozenset({"display_name", "target"}), ) @@ -344,30 +299,23 @@ def validate_roster( raise DiscussionValidationError(f"member {index} display_name is too long") target_key = json.dumps(target, sort_keys=True, separators=(",", ":")).casefold() - handle_key = handle.casefold() - member_key = member_id.casefold() - if target_key in targets: - if target.get("kind") == "local": - raise DiscussionValidationError("member profiles must be unique") - raise DiscussionValidationError("member targets must be unique") - if handle_key in handles or handle_key in {"all", "everyone"}: - raise DiscussionValidationError( - "member handles must be unique and cannot reserve @all or @everyone" - ) - if member_key in member_ids: - raise DiscussionValidationError("member ids must be unique") - targets.add(target_key) - handles.add(handle_key) - member_ids.add(member_key) - members.append( - DiscussionMember( - member_id=member_id, - profile=profile, - handle=handle, - display_name=display_name, - target=target, - ) + target_message = ( + "member profiles must be unique" if target.get("kind") == "local" + else "member targets must be unique" ) + 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"), + ): + 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, + )) return tuple(members) @@ -388,11 +336,8 @@ def validate_room(value: Any, *, local_profiles: Iterable[str]) -> DiscussionRoo 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, + room_id=room_id, name=name, members=members, + gateway_id=gateway_id, authority_epoch=authority_epoch, ) @@ -403,10 +348,7 @@ def is_pass_text(value: Any) -> bool: def resolve_mentions( - texts: Iterable[str], - members: Sequence[DiscussionMember], - *, - default_all: bool = True, + texts: Iterable[str], members: Sequence[DiscussionMember], *, default_all: bool = True ) -> tuple[DiscussionMember, ...]: """Resolve member handles deterministically against the frozen roster.""" by_handle = {member.handle.casefold(): member for member in members} @@ -440,8 +382,7 @@ def _unaddressed_member_mentions( if member.member_id != speaker_id: cited_at[member.member_id] = event.seq return tuple( - member - for member in room.members + member for member in room.members if member.member_id in cited_at and last_post_at.get(member.member_id, 0) <= cited_at[member.member_id] ) @@ -452,6 +393,114 @@ def _require_gateway_actor(actor: Mapping[str, Any], room: DiscussionRoom, messa raise DiscussionValidationError(message) +def _member_by_id(room: DiscussionRoom, member_id: Any) -> DiscussionMember: + normalized = _identifier(member_id, label="member_id") + for member in room.members: + if member.member_id == normalized: + return member + raise DiscussionValidationError(f"unknown Discussion member '{normalized}'") + + +def _validate_turn_coordinates(payload: Mapping[str, Any], room: DiscussionRoom) -> None: + _member_by_id(room, payload.get("member_id")) + _zero_based_int(payload.get("member_index"), label="member_index", maximum=MAX_DISCUSSION_MEMBERS - 1) + _zero_based_int(payload.get("round_index"), label="round_index", maximum=MAX_DISCUSSION_ROUNDS - 1) + for field in ("thread_id", "task_id", "turn_id", "discussion_event_id"): + _identifier(payload.get(field), label=field) + + +# -- per-kind event payload validators (dispatched by _validate_event) --------- +# 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]: + 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]: + _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_connection = peer.get("peer_id") if peer else None + if ( + actor.get("kind") != "member" + or actor.get("id") != member.member_id + or actor.get("profile") != member.profile + or actor.get("connection_id") != expected_connection + ): + 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]: + _exact_fields( + payload, label=f"{kind} payload", + required=_TERMINAL_COMMON_FIELDS | _TERMINAL_EXTRA_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") + if actor.get("connection_id") is not None: + raise DiscussionValidationError(f"{kind} requires a gateway actor") + _require_gateway_actor(actor, room, f"{kind} requires a gateway actor") + if kind == "turn.settled": + if not isinstance(payload.get("passed"), bool): + raise DiscussionValidationError("turn.settled passed must be a boolean") + message_event_id = payload.get("message_event_id") + if not payload["passed"]: + _identifier(message_event_id, label="message_event_id") + elif message_event_id is not None: + raise DiscussionValidationError("a passed turn cannot reference a member message") + return payload + field = "error" if kind == "turn.failed" else "reason" + if not isinstance(payload.get(field), str) or not payload[field].strip(): + 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" + ) + return payload + + +def _validate_gateway_event( + kind: str, payload: Mapping[str, Any], actor: Mapping[str, Any], room: DiscussionRoom +) -> Mapping[str, Any]: + 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"}: + raise DiscussionValidationError("invalid room.activity status") + for field in identifier_fields: + _identifier(payload.get(field), label=field) + _require_gateway_actor(actor, room, f"{kind} requires the room gateway") + return payload + + +_EVENT_VALIDATORS = { + "message.user": _validate_user_event, + "message.member": _validate_member_message, + **dict.fromkeys(_TERMINAL_EVENT_KINDS, _validate_terminal_event), + **dict.fromkeys(_GATEWAY_EVENT_FIELDS, _validate_gateway_event), +} + + def _validate_event(raw: Any, *, room: DiscussionRoom, previous_seq: int) -> _ValidatedEvent: if not isinstance(raw, Mapping): raise DiscussionValidationError("room event must be an object") @@ -472,104 +521,10 @@ def _validate_event(raw: Any, *, room: DiscussionRoom, previous_seq: int) -> _Va raise DiscussionValidationError("event payload must be an object") 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") - - if kind == "message.user": - payload = validate_user_payload(payload) - if actor.get("kind") != "user": - raise DiscussionValidationError("message.user requires a user actor") - elif kind == "message.member": - _validate_member_message(payload, actor=actor, room=room) - elif kind in _TERMINAL_EVENT_KINDS: - _validate_terminal_event(kind, payload, actor=actor, room=room) - elif kind in _GATEWAY_EVENT_FIELDS: - 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"}: - raise DiscussionValidationError("invalid room.activity status") - for field in identifier_fields: - _identifier(payload.get(field), label=field) - _require_gateway_actor(actor, room, f"{kind} requires the room gateway") - - return _ValidatedEvent( - raw=raw, seq=seq, event_id=event_id, kind=kind, actor=actor, payload=payload - ) - - -def _member_by_id(room: DiscussionRoom, member_id: Any) -> DiscussionMember: - normalized = _identifier(member_id, label="member_id") - for member in room.members: - if member.member_id == normalized: - return member - raise DiscussionValidationError(f"unknown Discussion member '{normalized}'") - - -def _validate_turn_coordinates(payload: Mapping[str, Any], room: DiscussionRoom) -> None: - _member_by_id(room, payload.get("member_id")) - _zero_based_int( - payload.get("member_index"), label="member_index", maximum=MAX_DISCUSSION_MEMBERS - 1 - ) - _zero_based_int( - payload.get("round_index"), label="round_index", maximum=MAX_DISCUSSION_ROUNDS - 1 - ) - for field in ("thread_id", "task_id", "turn_id", "discussion_event_id"): - _identifier(payload.get(field), label=field) - - -def _validate_member_message( - payload: Mapping[str, Any], *, actor: Mapping[str, Any], room: DiscussionRoom -) -> None: - _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_connection = peer.get("peer_id") if peer else None - if ( - actor.get("kind") != "member" - or actor.get("id") != member.member_id - or actor.get("profile") != member.profile - or actor.get("connection_id") != expected_connection - ): - raise DiscussionValidationError("message.member actor does not match roster") - - -def _validate_terminal_event( - kind: str, payload: Mapping[str, Any], *, actor: Mapping[str, Any], room: DiscussionRoom -) -> None: - _exact_fields( - payload, - label=f"{kind} payload", - required=_TERMINAL_COMMON_FIELDS | _TERMINAL_EXTRA_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") - if actor.get("connection_id") is not None: - raise DiscussionValidationError(f"{kind} requires a gateway actor") - _require_gateway_actor(actor, room, f"{kind} requires a gateway actor") - if kind == "turn.settled": - if not isinstance(payload.get("passed"), bool): - raise DiscussionValidationError("turn.settled passed must be a boolean") - message_event_id = payload.get("message_event_id") - if not payload["passed"]: - _identifier(message_event_id, label="message_event_id") - elif message_event_id is not None: - raise DiscussionValidationError("a passed turn cannot reference a member message") - return - field = "error" if kind == "turn.failed" else "reason" - if not isinstance(payload.get(field), str) or not payload[field].strip(): - 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" - ) + validator = _EVENT_VALIDATORS.get(kind) + if validator 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( @@ -589,22 +544,15 @@ def _validated_events( def derive_member_watermarks( - room_value: Any, - events: Sequence[Mapping[str, Any]], - *, - local_profiles: Iterable[str], + 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)) -def _derive_member_watermarks( - events: Sequence[_ValidatedEvent], -) -> dict[tuple[str, str], int]: - messages_by_id = { - event.event_id: event for event in events if event.kind == "message.member" - } +def _derive_member_watermarks(events: Sequence[_ValidatedEvent]) -> dict[tuple[str, str], int]: + messages_by_id = {event.event_id: event for event in events if event.kind == "message.member"} terminal_by_task: dict[str, _ValidatedEvent] = {} watermarks: dict[tuple[str, str], int] = {} for event in events: @@ -634,9 +582,7 @@ def _derive_member_watermarks( or message.payload.get("member_id") != event.payload.get("member_id") or message.payload.get("thread_id") != event.payload.get("thread_id") ): - raise DiscussionValidationError( - "turn.settled references no matching member message" - ) + raise DiscussionValidationError("turn.settled references no matching member message") watermark = max(watermark, message.seq) watermarks[key] = max(watermarks.get(key, 0), watermark) return watermarks @@ -645,8 +591,7 @@ def _derive_member_watermarks( def _member_digest(member: DiscussionMember) -> str: target = json.dumps( member.target or {"kind": "local", "profile": member.profile}, - sort_keys=True, - separators=(",", ":"), + sort_keys=True, separators=(",", ":"), ) seed = f"{member.member_id}\0{member.profile}\0{member.handle}\0{target}" return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24] @@ -683,20 +628,14 @@ def _truncate_utf8_text(value: Any, *, max_bytes: int, suffix: str = "") -> str: def _build_prompt( - *, - room: DiscussionRoom, - member: DiscussionMember, - messages: Sequence[_ValidatedEvent], - watermark: int, - seen_through_seq: int, + *, room: DiscussionRoom, member: DiscussionMember, messages: Sequence[_ValidatedEvent], + watermark: int, seen_through_seq: int, ) -> str: delta = [event for event in messages if watermark < event.seq <= seen_through_seq][ -MAX_DISCUSSION_DELTA_LINES: ] peers = ", ".join( - f"@{candidate.handle}" - for candidate in room.members - if candidate.member_id != member.member_id + f"@{candidate.handle}" for candidate in room.members if candidate.member_id != member.member_id ) opening = [ f'[Discussion: "{room.name}"] You are @{member.handle}, one participant ' @@ -737,14 +676,8 @@ 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, + *, 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}." @@ -762,16 +695,12 @@ def _make_task_plan( "source_event_seq": discussion_event.seq, "thread_id": discussion_event.payload["thread_id"], }, - ensure_ascii=True, - sort_keys=True, - separators=(",", ":"), + ensure_ascii=True, sort_keys=True, separators=(",", ":"), ) 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=task_id, + thread_id=str(discussion_event.payload["thread_id"]), turn_id=turn_id, ) payload = { "target_member_id": member.member_id, @@ -780,69 +709,43 @@ def _make_task_plan( "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, + 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, ) -def plan_next_task( - room_value: Any, - events: Sequence[Mapping[str, Any]], - *, - local_profiles: Iterable[str], - initial_watermarks: Mapping[tuple[str, str], int] | None = None, -) -> DiscussionDecision: - """Replay the complete room log and return at most one next member task.""" - room = validate_room(room_value, local_profiles=local_profiles) - validated = _validated_events(events, room=room) +def _pending_discussion(validated: Sequence[_ValidatedEvent]) -> _ValidatedEvent | None: + """Oldest latest-per-thread user message not stopped and not yet completed.""" stopped_through_seq = max( - (event.seq for event in validated if event.kind == "room.stop_requested"), - default=0, + (event.seq for event in validated if event.kind == "room.stop_requested"), default=0 ) completed_discussion_ids = { str(event.payload["discussion_event_id"]) for event in validated - if event.kind == "room.activity" - and event.payload.get("status") in {"settled", "bounded"} + 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_user_events = tuple( + 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 - ) - if not pending_user_events: - return DiscussionDecision(status="idle", reason="no_pending_user_event") + if event.seq > stopped_through_seq and event.event_id not in completed_discussion_ids + ] + return pending[0] if pending else None - discussion = pending_user_events[0] + +def _thread_messages( + validated: Sequence[_ValidatedEvent], discussion: _ValidatedEvent +) -> tuple[tuple[_ValidatedEvent, ...], tuple[_ValidatedEvent, ...], tuple[_ValidatedEvent, ...]]: + """Return (thread messages, messages since the discussion, this discussion's member messages).""" thread_id = str(discussion.payload["thread_id"]) - - def decide( - status: DecisionStatus, reason: str, task: DiscussionTaskPlan | None = None - ) -> DiscussionDecision: - return DiscussionDecision( - status=status, - reason=reason, - discussion_event_id=discussion.event_id, - source_event_seq=discussion.seq, - thread_id=thread_id, - task=task, - ) - committed_member_message_ids = { str(event.payload["message_event_id"]) for event in validated - if event.kind == "turn.settled" - and event.payload.get("message_event_id") is not None + if event.kind == "turn.settled" and event.payload.get("message_event_id") is not None } # Publication writes the visible member message before the terminal event. # A crash in that gap leaves the message in the log, but it is not committed @@ -857,15 +760,48 @@ def plan_next_task( or (event.kind == "message.member" and event.event_id in committed_member_message_ids) ) ) - discussion_messages = tuple( - event for event in thread_messages if event.seq >= discussion.seq - ) + discussion_messages = tuple(event for event in thread_messages if event.seq >= discussion.seq) member_messages = tuple( - event - for event in thread_messages + event for event in thread_messages if event.kind == "message.member" and event.payload.get("discussion_event_id") == discussion.event_id ) + return thread_messages, discussion_messages, member_messages + + +def _effective_watermarks( + validated: Sequence[_ValidatedEvent], initial_watermarks: Mapping[tuple[str, str], int] | None +) -> 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 key, value in _derive_member_watermarks(validated).items(): + watermarks[key] = max(watermarks.get(key, 0), value) + return watermarks + + +def plan_next_task( + room_value: Any, + events: Sequence[Mapping[str, Any]], + *, + local_profiles: Iterable[str], + initial_watermarks: Mapping[tuple[str, str], int] | None = None, +) -> DiscussionDecision: + """Replay the complete room log and return at most one next member task.""" + room = validate_room(room_value, local_profiles=local_profiles) + validated = _validated_events(events, room=room) + discussion = _pending_discussion(validated) + if discussion is None: + return DiscussionDecision(status="idle", reason="no_pending_user_event") + thread_id = str(discussion.payload["thread_id"]) + decide = partial( + DiscussionDecision, discussion_event_id=discussion.event_id, + source_event_seq=discussion.seq, thread_id=thread_id, + ) + + thread_messages, discussion_messages, member_messages = _thread_messages(validated, discussion) if len(member_messages) >= MAX_DISCUSSION_MESSAGES: return decide("bounded", "max_messages") @@ -875,13 +811,7 @@ def plan_next_task( if event.kind in _TERMINAL_EVENT_KINDS and event.payload.get("discussion_event_id") == discussion.event_id } - 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 key, value in _derive_member_watermarks(validated).items(): - watermarks[key] = max(watermarks.get(key, 0), value) + watermarks = _effective_watermarks(validated, initial_watermarks) seen_through_seq = max(event.seq for event in thread_messages) for round_index in range(MAX_DISCUSSION_ROUNDS): @@ -902,27 +832,16 @@ def plan_next_task( if not any(watermark < event.seq <= seen_through_seq for event in thread_messages): continue prompt = _build_prompt( - room=room, - member=member, - messages=thread_messages, - watermark=watermark, - seen_through_seq=seen_through_seq, + room=room, member=member, messages=thread_messages, + watermark=watermark, seen_through_seq=seen_through_seq, ) task = _make_task_plan( - room=room, - discussion_event=discussion, - member=member, - member_index=member_index, - round_index=round_index, - seen_through_seq=seen_through_seq, - prompt=prompt, + room=room, discussion_event=discussion, member=member, member_index=member_index, + round_index=round_index, seen_through_seq=seen_through_seq, prompt=prompt, ) - return decide("task", "member_turn", task) + return decide("task", "member_turn", task=task) - spoke = any( - int(event.payload["round_index"]) == round_index for event in member_messages - ) - if not spoke: + if not any(int(event.payload["round_index"]) == round_index for event in member_messages): return decide("settled", "silent_round") if round_index == MAX_DISCUSSION_ROUNDS - 1: return decide("bounded", "max_rounds") @@ -931,11 +850,8 @@ def plan_next_task( def reconstruct_task_plan( - room_value: Any, - events: Sequence[Mapping[str, Any]], - task: Mapping[str, Any], - *, - local_profiles: Iterable[str], + room_value: Any, events: Sequence[Mapping[str, Any]], task: Mapping[str, Any], + *, local_profiles: Iterable[str], ) -> DiscussionTaskPlan: """Reconstruct and verify one persisted driver task after a restart.""" room = validate_room(room_value, local_profiles=local_profiles) @@ -956,12 +872,7 @@ def reconstruct_task_plan( if payload.get("source_event_seq") != source_event_seq: raise DiscussionReconstructionError("task source event does not match turn_id") discussion = next( - ( - event - for event in validated - if event.seq == source_event_seq and event.kind == "message.user" - ), - None, + (e for e in validated if e.seq == source_event_seq and e.kind == "message.user"), None ) if discussion is None: raise DiscussionReconstructionError("task source user event is missing") @@ -971,8 +882,7 @@ def reconstruct_task_plan( target_member_id = payload.get("target_member_id") member = next( ( - candidate - for candidate in room.members + candidate for candidate in room.members if ( candidate.member_id == target_member_id if target_member_id is not None @@ -991,13 +901,9 @@ def reconstruct_task_plan( if len(prompt.encode("utf-8")) > driver.MAX_PROMPT_BYTES: raise DiscussionReconstructionError("task prompt exceeds the driver limit") reconstructed = _make_task_plan( - room=room, - discussion_event=discussion, - member=member, - member_index=int(match.group("position")), - round_index=int(match.group("round")), - seen_through_seq=int(match.group("seen")), - prompt=prompt, + room=room, discussion_event=discussion, member=member, + member_index=int(match.group("position")), round_index=int(match.group("round")), + seen_through_seq=int(match.group("seen")), prompt=prompt, ) if reconstructed.identity != identity or dict(reconstructed.payload) != dict(payload): raise DiscussionReconstructionError("driver task failed deterministic reconstruction") @@ -1015,6 +921,82 @@ def _terminal_text(result: Any, *, field: str, fallback: str) -> str: return text or fallback +# -- per-status terminal payload builders (dispatched by plan_publication) ----- +# Each returns (extra terminal payload fields, visible effects to publish first). + + +def _settled_effects( + result: Any, *, task: DiscussionTaskPlan, room: DiscussionRoom, message_event_id: str, **_: Any +) -> tuple[dict[str, Any], list[EventPlan]]: + 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={ + "discussion_event_id": task.discussion_event_id, + "member_id": task.member.member_id, + "member_index": task.member_index, + "round_index": task.round_index, + "task_id": task.identity.task_id, + "text": text, + "thread_id": task.identity.thread_id, + "turn_id": task.identity.turn_id, + }, + 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 + + +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 + + 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) + 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 _deferred_effects( + result: Any, *, execution_generation: int | None, **_: Any +) -> tuple[dict[str, Any], list[EventPlan]]: + return { + "execution_generation": execution_generation, + "reason": _terminal_text(result, field="reason", fallback="member_unavailable"), + }, [] + + +_TERMINAL_EFFECTS = { + "settled": _settled_effects, + "failed": _failed_effects, + "cancelled": _cancelled_effects, + "deferred": _deferred_effects, +} + + def plan_publication( room_value: Any, events: Sequence[Mapping[str, Any]], @@ -1037,7 +1019,7 @@ def plan_publication( raise DiscussionValidationError("task belongs to a different room") if task.member not in room.members: raise DiscussionValidationError("task member is not in the frozen roster") - if status not in {"settled", "failed", "cancelled", "deferred"}: + if status not in _TERMINAL_EFFECTS: raise DiscussionValidationError("invalid terminal publication status") if status == "deferred" and ( isinstance(execution_generation, bool) @@ -1056,13 +1038,11 @@ def plan_publication( "cancelled" if newer_same_thread and status != "deferred" else status ) digest = task.identity.task_id.removeprefix("dtask:") - message_event_id = f"dmessage:{digest}" terminal_event_id = ( f"ddeferred:{digest}:g{execution_generation}" - if effective_status == "deferred" - else f"dterminal:{digest}" + if effective_status == "deferred" else f"dterminal:{digest}" ) - common = { + common_payload = { "discussion_event_id": task.discussion_event_id, "member_id": task.member.member_id, "member_index": task.member_index, @@ -1072,86 +1052,17 @@ def plan_publication( "thread_id": task.identity.thread_id, "turn_id": task.identity.turn_id, } - effects: list[EventPlan] = [] + 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}" - - if effective_status == "settled": - 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) - 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={ - "discussion_event_id": task.discussion_event_id, - "member_id": task.member.member_id, - "member_index": task.member_index, - "round_index": task.round_index, - "task_id": task.identity.task_id, - "text": text, - "thread_id": task.identity.thread_id, - "turn_id": task.identity.turn_id, - }, - authority_gateway_id=room.gateway_id, - authority_epoch=room.authority_epoch, - ) - ) - extra = {"message_event_id": None if passed else message_event_id, "passed": passed} - elif effective_status == "failed": - error_text = _terminal_text(result, field="error", fallback="member turn failed") - from tools.bot_failure_reasons import ALL_REASONS, 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) - ) - extra = {"error": error_text, "reason_code": reason_code} - elif effective_status == "cancelled": - extra = { - "reason": ( - "superseded_by_newer_user_event" - if newer_same_thread - else _terminal_text(result, field="reason", fallback="member turn cancelled") - ), - } - else: - extra = { - "execution_generation": execution_generation, - "reason": _terminal_text(result, field="reason", fallback="member_unavailable"), - } - - effects.append( - EventPlan( - event_id=terminal_event_id, - kind=terminal_kind, - actor={"kind": "gateway", "id": room.gateway_id}, - payload={**common, **extra}, - authority_gateway_id=room.gateway_id, - authority_epoch=room.authority_epoch, - ) - ) + 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, + )) return PublicationPlan( - task_id=task.identity.task_id, - terminal_kind=terminal_kind, - events=tuple(effects), + task_id=task.identity.task_id, terminal_kind=terminal_kind, events=tuple(effects) ) diff --git a/gateway/hosted_room_policy_checkpoint.py b/gateway/hosted_room_policy_checkpoint.py index ab1234aa4d..590f109ed8 100644 --- a/gateway/hosted_room_policy_checkpoint.py +++ b/gateway/hosted_room_policy_checkpoint.py @@ -19,75 +19,55 @@ from gateway import hosted_rooms MAX_ACTIVE_POLICY_EVENTS = 64 MAX_THREAD_TRANSCRIPT_EVENTS = 24 _TRANSCRIPT_SCHEMA_VERSION = 1 -_TERMINAL_KINDS = frozenset({ - "turn.settled", - "turn.failed", - "turn.cancelled", - "turn.deferred", -}) +_TERMINAL_KINDS = frozenset({"turn.settled", "turn.failed", "turn.cancelled", "turn.deferred"}) _SCHEMA_DDL = ( """CREATE TABLE IF NOT EXISTS hosted_room_policy_cursors ( - room_id TEXT PRIMARY KEY, - through_seq INTEGER NOT NULL DEFAULT 0, - stopped_through_seq INTEGER NOT NULL DEFAULT 0, - updated_at REAL NOT NULL DEFAULT 0 - )""", + room_id TEXT PRIMARY KEY, through_seq INTEGER NOT NULL DEFAULT 0, + stopped_through_seq INTEGER NOT NULL DEFAULT 0, updated_at REAL NOT NULL DEFAULT 0)""", """CREATE TABLE IF NOT EXISTS hosted_room_policy_threads ( - room_id TEXT NOT NULL, - thread_id TEXT NOT NULL, - discussion_event_id TEXT NOT NULL, - latest_user_seq INTEGER NOT NULL, - completed INTEGER NOT NULL DEFAULT 0, - PRIMARY KEY(room_id, thread_id) - )""", + room_id TEXT NOT NULL, thread_id TEXT NOT NULL, discussion_event_id TEXT NOT NULL, + latest_user_seq INTEGER NOT NULL, completed INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY(room_id, thread_id))""", """CREATE INDEX IF NOT EXISTS idx_hosted_room_policy_pending - ON hosted_room_policy_threads( - room_id, completed, latest_user_seq, thread_id - )""", + ON hosted_room_policy_threads(room_id, completed, latest_user_seq, thread_id)""", """CREATE TABLE IF NOT EXISTS hosted_room_policy_events ( - room_id TEXT NOT NULL, - thread_id TEXT NOT NULL, - discussion_event_id TEXT NOT NULL, - seq INTEGER NOT NULL, - event_json TEXT NOT NULL, - PRIMARY KEY(room_id, seq) - )""", + room_id TEXT NOT NULL, thread_id TEXT NOT NULL, discussion_event_id TEXT NOT NULL, + seq INTEGER NOT NULL, event_json TEXT NOT NULL, PRIMARY KEY(room_id, seq))""", """CREATE INDEX IF NOT EXISTS idx_hosted_room_policy_events_active - ON hosted_room_policy_events( - room_id, discussion_event_id, seq - )""", + ON hosted_room_policy_events(room_id, discussion_event_id, seq)""", """CREATE TABLE IF NOT EXISTS hosted_room_policy_watermarks ( - room_id TEXT NOT NULL, - thread_id TEXT NOT NULL, - member_id TEXT NOT NULL, - seen_through_seq INTEGER NOT NULL, - PRIMARY KEY(room_id, thread_id, member_id) - )""", + room_id TEXT NOT NULL, thread_id TEXT NOT NULL, member_id TEXT NOT NULL, + seen_through_seq INTEGER NOT NULL, PRIMARY KEY(room_id, thread_id, member_id))""", """CREATE TABLE IF NOT EXISTS hosted_room_policy_publications ( - room_id TEXT NOT NULL, - task_id TEXT NOT NULL, - kind TEXT NOT NULL, - execution_generation INTEGER NOT NULL DEFAULT 0, - seq INTEGER NOT NULL, - PRIMARY KEY(room_id, task_id, kind, execution_generation) - )""", + room_id TEXT NOT NULL, task_id TEXT NOT NULL, kind TEXT NOT NULL, + execution_generation INTEGER NOT NULL DEFAULT 0, seq INTEGER NOT NULL, + PRIMARY KEY(room_id, task_id, kind, execution_generation))""", # Transcript stores only references into the already bounded room log, so # prompt payloads are never duplicated outside room byte limits. """CREATE TABLE IF NOT EXISTS hosted_room_policy_transcript ( - room_id TEXT NOT NULL, - thread_id TEXT NOT NULL, - seq INTEGER NOT NULL, - kind TEXT NOT NULL, - settled_seq INTEGER, - PRIMARY KEY(room_id, thread_id, seq) - )""", + room_id TEXT NOT NULL, thread_id TEXT NOT NULL, seq INTEGER NOT NULL, + kind TEXT NOT NULL, settled_seq INTEGER, PRIMARY KEY(room_id, thread_id, seq))""", """CREATE TABLE IF NOT EXISTS hosted_room_policy_transcript_state ( - room_id TEXT PRIMARY KEY, - schema_version INTEGER NOT NULL - )""", + room_id TEXT PRIMARY KEY, schema_version INTEGER NOT NULL)""", ) +_ROOM_EVENT_COLUMNS = ( + "room_id, seq, event_id, kind, actor_json, authority_epoch, payload_json, created_at" +) +_DELETE_ACTIVE_EVENTS_SQL = ( + "DELETE FROM hosted_room_policy_events WHERE room_id=? AND discussion_event_id=?" +) +_TRANSCRIPT_EVENTS_SQL = f"""WITH transcript_events(seq) AS ( + SELECT seq FROM hosted_room_policy_transcript WHERE room_id=? AND thread_id=? + UNION ALL + SELECT settled_seq FROM hosted_room_policy_transcript + WHERE room_id=? AND thread_id=? AND settled_seq IS NOT NULL) + SELECT {", ".join("events." + column for column in _ROOM_EVENT_COLUMNS.split(", "))} + FROM transcript_events + JOIN hosted_room_events AS events ON events.room_id=? AND events.seq=transcript_events.seq + ORDER BY events.seq""" + @dataclass(frozen=True) class PolicySnapshot: @@ -101,23 +81,15 @@ class PolicySnapshot: def _event_from_room_row(row: sqlite3.Row) -> dict[str, Any]: return { - "room_id": str(row["room_id"]), - "seq": int(row["seq"]), - "event_id": str(row["event_id"]), - "kind": str(row["kind"]), - "actor": json.loads(row["actor_json"]), - "authority_epoch": row["authority_epoch"], - "payload": json.loads(row["payload_json"]), - "created_at": float(row["created_at"]), - "idempotent": False, + "room_id": str(row["room_id"]), "seq": int(row["seq"]), "event_id": str(row["event_id"]), + "kind": str(row["kind"]), "actor": json.loads(row["actor_json"]), + "authority_epoch": row["authority_epoch"], "payload": json.loads(row["payload_json"]), + "created_at": float(row["created_at"]), "idempotent": False, } def _require_room(conn: sqlite3.Connection, room_id: str) -> None: - if conn.execute( - "SELECT 1 FROM hosted_rooms WHERE room_id=?", - (room_id,), - ).fetchone() is None: + if conn.execute("SELECT 1 FROM hosted_rooms WHERE room_id=?", (room_id,)).fetchone() is None: raise hosted_rooms.RoomNotFoundError("hosted room not found") @@ -141,36 +113,21 @@ class HostedRoomPolicyCheckpoint: @staticmethod def _store_active_event( - conn: sqlite3.Connection, - *, - event: Mapping[str, Any], - thread_id: str, - discussion_event_id: str, + conn: sqlite3.Connection, *, event: Mapping[str, Any], thread_id: str, discussion_event_id: str ) -> None: conn.execute( """INSERT OR IGNORE INTO hosted_room_policy_events( room_id, thread_id, discussion_event_id, seq, event_json ) VALUES (?, ?, ?, ?, ?)""", ( - event["room_id"], - thread_id, - discussion_event_id, - int(event["seq"]), - json.dumps( - dict(event), - ensure_ascii=True, - sort_keys=True, - separators=(",", ":"), - ), + event["room_id"], thread_id, discussion_event_id, int(event["seq"]), + json.dumps(dict(event), ensure_ascii=True, sort_keys=True, separators=(",", ":")), ), ) @staticmethod def _store_transcript_event( - conn: sqlite3.Connection, - *, - event: Mapping[str, Any], - thread_id: str, + conn: sqlite3.Connection, *, event: Mapping[str, Any], thread_id: str, settled_seq: int | None = None, ) -> None: conn.execute( @@ -178,125 +135,66 @@ class HostedRoomPolicyCheckpoint: room_id, thread_id, seq, kind, settled_seq ) VALUES (?, ?, ?, ?, ?) ON CONFLICT(room_id, thread_id, seq) DO UPDATE SET - settled_seq=COALESCE( - excluded.settled_seq, - hosted_room_policy_transcript.settled_seq - )""", - ( - event["room_id"], - thread_id, - int(event["seq"]), - str(event["kind"]), - settled_seq, - ), + settled_seq=COALESCE(excluded.settled_seq, hosted_room_policy_transcript.settled_seq)""", + (event["room_id"], thread_id, int(event["seq"]), str(event["kind"]), settled_seq), ) if event["kind"] in {"message.user", "message.member"}: cutoff = conn.execute( """SELECT seq FROM hosted_room_policy_transcript - WHERE room_id=? AND thread_id=? - AND kind IN ('message.user', 'message.member') + WHERE room_id=? AND thread_id=? AND kind IN ('message.user', 'message.member') ORDER BY seq DESC LIMIT 1 OFFSET ?""", - ( - event["room_id"], - thread_id, - MAX_THREAD_TRANSCRIPT_EVENTS - 1, - ), + (event["room_id"], thread_id, MAX_THREAD_TRANSCRIPT_EVENTS - 1), ).fetchone() if cutoff is not None: conn.execute( - """DELETE FROM hosted_room_policy_transcript - WHERE room_id=? AND thread_id=? AND seq None: + def _backfill_transcript(self, conn: sqlite3.Connection, *, room_id: str, through_seq: int) -> None: """Migrate bounded committed thread history from the durable room log.""" - if through_seq <= 0: return settled_seq_by_message: dict[str, int] = {} for row in conn.execute( """SELECT seq, payload_json FROM hosted_room_events - WHERE room_id=? AND seq<=? AND kind='turn.settled' - ORDER BY seq""", + WHERE room_id=? AND seq<=? AND kind='turn.settled' ORDER BY seq""", (room_id, through_seq), ): - message_event_id = str( - json.loads(row["payload_json"]).get("message_event_id") or "" - ) + message_event_id = str(json.loads(row["payload_json"]).get("message_event_id") or "") if message_event_id: settled_seq_by_message[message_event_id] = int(row["seq"]) rows = conn.execute( - """SELECT room_id, seq, event_id, kind, actor_json, - authority_epoch, payload_json, created_at - FROM hosted_room_events - WHERE room_id=? AND seq<=? - AND kind IN ('message.user', 'message.member') + f"""SELECT {_ROOM_EVENT_COLUMNS} FROM hosted_room_events + WHERE room_id=? AND seq<=? AND kind IN ('message.user', 'message.member') ORDER BY seq""", (room_id, through_seq), ) for row in rows: - if ( - row["kind"] == "message.member" - and row["event_id"] not in settled_seq_by_message - ): + if row["kind"] == "message.member" and row["event_id"] not in settled_seq_by_message: continue event = _event_from_room_row(row) thread_id = str(event["payload"].get("thread_id") or "") if thread_id: self._store_transcript_event( - conn, - event=event, - thread_id=thread_id, + conn, event=event, thread_id=thread_id, settled_seq=settled_seq_by_message.get(str(row["event_id"])), ) def _discussion_events( - self, - conn: sqlite3.Connection, - *, - room_id: str, - thread_id: str, - discussion_event_id: str, + self, conn: sqlite3.Connection, *, room_id: str, thread_id: str, discussion_event_id: str, bound_error: str, ) -> list[dict[str, Any]]: """Merge the thread transcript with the active projection, ordered by seq.""" - active_rows = conn.execute( """SELECT event_json FROM hosted_room_policy_events - WHERE room_id=? AND discussion_event_id=? - ORDER BY seq LIMIT ?""", - ( - room_id, - discussion_event_id, - MAX_ACTIVE_POLICY_EVENTS + 1, - ), + WHERE room_id=? AND discussion_event_id=? ORDER BY seq LIMIT ?""", + (room_id, discussion_event_id, MAX_ACTIVE_POLICY_EVENTS + 1), ).fetchall() if len(active_rows) > MAX_ACTIVE_POLICY_EVENTS: raise RuntimeError(bound_error) rows = conn.execute( - """WITH transcript_events(seq) AS ( - SELECT seq FROM hosted_room_policy_transcript - WHERE room_id=? AND thread_id=? - UNION ALL - SELECT settled_seq FROM hosted_room_policy_transcript - WHERE room_id=? AND thread_id=? AND settled_seq IS NOT NULL - ) - SELECT events.room_id, events.seq, events.event_id, - events.kind, events.actor_json, - events.authority_epoch, events.payload_json, - events.created_at - FROM transcript_events - JOIN hosted_room_events AS events - ON events.room_id=? AND events.seq=transcript_events.seq - ORDER BY events.seq""", - (room_id, thread_id, room_id, thread_id, room_id), + _TRANSCRIPT_EVENTS_SQL, (room_id, thread_id, room_id, thread_id, room_id) ).fetchall() events_by_seq = { int(event["seq"]): event @@ -310,10 +208,7 @@ class HostedRoomPolicyCheckpoint: # -- per-kind projection handlers (dispatched by _apply_event) ----------- def _apply_user_message( - self, - conn: sqlite3.Connection, - event: Mapping[str, Any], - payload: Mapping[str, Any], + self, conn: sqlite3.Connection, event: Mapping[str, Any], payload: Mapping[str, Any] ) -> None: room_id = str(event["room_id"]) thread_id = str(payload.get("thread_id") or "") @@ -322,84 +217,57 @@ class HostedRoomPolicyCheckpoint: return conn.execute( """INSERT INTO hosted_room_policy_threads( - room_id, thread_id, discussion_event_id, - latest_user_seq, completed - ) VALUES (?, ?, ?, ?, 0) - ON CONFLICT(room_id, thread_id) DO UPDATE SET - discussion_event_id=excluded.discussion_event_id, - latest_user_seq=excluded.latest_user_seq, - completed=0""", + room_id, thread_id, discussion_event_id, latest_user_seq, completed + ) VALUES (?, ?, ?, ?, 0) + ON CONFLICT(room_id, thread_id) DO UPDATE SET + discussion_event_id=excluded.discussion_event_id, + latest_user_seq=excluded.latest_user_seq, completed=0""", (room_id, thread_id, event_id, int(event["seq"])), ) - self._store_active_event( - conn, - event=event, - thread_id=thread_id, - discussion_event_id=event_id, - ) + self._store_active_event(conn, event=event, thread_id=thread_id, discussion_event_id=event_id) self._store_transcript_event(conn, event=event, thread_id=thread_id) def _apply_discussion_event( - self, - conn: sqlite3.Connection, - event: Mapping[str, Any], - payload: Mapping[str, Any], + self, conn: sqlite3.Connection, event: Mapping[str, Any], payload: Mapping[str, Any] ) -> None: """Index member messages and terminal turn outcomes of a known discussion.""" - room_id = str(event["room_id"]) seq = int(event["seq"]) kind = str(event.get("kind") or "") thread_id = str(payload.get("thread_id") or "") discussion_event_id = str(payload.get("discussion_event_id") or "") source = conn.execute( - """SELECT 1 FROM hosted_room_policy_events - WHERE room_id=? AND discussion_event_id=? LIMIT 1""", + "SELECT 1 FROM hosted_room_policy_events WHERE room_id=? AND discussion_event_id=? LIMIT 1", (room_id, discussion_event_id), ).fetchone() if source is None: return self._store_active_event( - conn, - event=event, - thread_id=thread_id, - discussion_event_id=discussion_event_id, + conn, event=event, thread_id=thread_id, discussion_event_id=discussion_event_id ) if kind not in _TERMINAL_KINDS: return task_id = str(payload.get("task_id") or "") execution_generation = ( - int(payload.get("execution_generation") or 0) - if kind == "turn.deferred" - else 0 + int(payload.get("execution_generation") or 0) if kind == "turn.deferred" else 0 ) if task_id: conn.execute( """INSERT OR IGNORE INTO hosted_room_policy_publications( - room_id, task_id, kind, execution_generation, seq - ) VALUES (?, ?, ?, ?, ?)""", - ( - room_id, - task_id, - kind, - execution_generation, - seq, - ), + room_id, task_id, kind, execution_generation, seq + ) VALUES (?, ?, ?, ?, ?)""", + (room_id, task_id, kind, execution_generation, seq), ) member_id = str(payload.get("member_id") or "") seen_through_seq = int(payload.get("seen_through_seq") or 0) if kind == "turn.settled" and payload.get("message_event_id"): messages = conn.execute( - """SELECT seq, event_json FROM hosted_room_policy_events - WHERE room_id=? AND discussion_event_id=?""", + "SELECT seq, event_json FROM hosted_room_policy_events WHERE room_id=? AND discussion_event_id=?", (room_id, discussion_event_id), ).fetchall() committed = next( ( - message - for message in ( - json.loads(row["event_json"]) for row in messages - ) + message for message in (json.loads(row["event_json"]) for row in messages) if message.get("event_id") == payload["message_event_id"] ), None, @@ -407,54 +275,36 @@ class HostedRoomPolicyCheckpoint: if committed is not None: seen_through_seq = max(seen_through_seq, int(committed["seq"])) self._store_transcript_event( - conn, - event=committed, - thread_id=thread_id, - settled_seq=int(event["seq"]), + conn, event=committed, thread_id=thread_id, settled_seq=int(event["seq"]) ) if member_id and seen_through_seq > 0: conn.execute( """INSERT INTO hosted_room_policy_watermarks( - room_id, thread_id, member_id, seen_through_seq - ) VALUES (?, ?, ?, ?) - ON CONFLICT(room_id, thread_id, member_id) DO UPDATE SET - seen_through_seq=MAX( - hosted_room_policy_watermarks.seen_through_seq, - excluded.seen_through_seq - )""", + room_id, thread_id, member_id, seen_through_seq + ) VALUES (?, ?, ?, ?) + ON CONFLICT(room_id, thread_id, member_id) DO UPDATE SET + seen_through_seq=MAX(hosted_room_policy_watermarks.seen_through_seq, excluded.seen_through_seq)""", (room_id, thread_id, member_id, seen_through_seq), ) def _apply_room_activity( - self, - conn: sqlite3.Connection, - event: Mapping[str, Any], - payload: Mapping[str, Any], + self, conn: sqlite3.Connection, event: Mapping[str, Any], payload: Mapping[str, Any] ) -> None: room_id = str(event["room_id"]) thread_id = str(payload.get("thread_id") or "") discussion_event_id = str(payload.get("discussion_event_id") or "") + conn.execute(_DELETE_ACTIVE_EVENTS_SQL, (room_id, discussion_event_id)) conn.execute( - """DELETE FROM hosted_room_policy_events - WHERE room_id=? AND discussion_event_id=?""", - (room_id, discussion_event_id), - ) - conn.execute( - """DELETE FROM hosted_room_policy_threads - WHERE room_id=? AND thread_id=?""", + "DELETE FROM hosted_room_policy_threads WHERE room_id=? AND thread_id=?", (room_id, thread_id), ) def _apply_stop_requested( - self, - conn: sqlite3.Connection, - event: Mapping[str, Any], - payload: Mapping[str, Any], + self, conn: sqlite3.Connection, event: Mapping[str, Any], payload: Mapping[str, Any] ) -> None: conn.execute( """UPDATE hosted_room_policy_cursors - SET stopped_through_seq=MAX(stopped_through_seq, ?) - WHERE room_id=?""", + SET stopped_through_seq=MAX(stopped_through_seq, ?) WHERE room_id=?""", (int(event["seq"]), str(event["room_id"])), ) @@ -473,59 +323,46 @@ class HostedRoomPolicyCheckpoint: payload = event.get("payload") handler(self, conn, event, payload if isinstance(payload, Mapping) else {}) + def _ensure_cursor_and_transcript(self, conn: sqlite3.Connection, room_id: str) -> int: + """Create the room cursor if absent, backfill the transcript once, return through_seq.""" + _require_room(conn, room_id) + conn.execute( + """INSERT OR IGNORE INTO hosted_room_policy_cursors( + room_id, through_seq, stopped_through_seq, updated_at + ) VALUES (?, 0, 0, 0)""", + (room_id,), + ) + row = conn.execute( + "SELECT through_seq FROM hosted_room_policy_cursors WHERE room_id=?", (room_id,) + ).fetchone() + cursor = int(row["through_seq"]) + transcript_state = conn.execute( + "SELECT schema_version FROM hosted_room_policy_transcript_state WHERE room_id=?", + (room_id,), + ).fetchone() + if transcript_state is None or int(transcript_state["schema_version"]) < _TRANSCRIPT_SCHEMA_VERSION: + self._backfill_transcript(conn, room_id=room_id, through_seq=cursor) + conn.execute( + """INSERT INTO hosted_room_policy_transcript_state(room_id, schema_version) + VALUES (?, ?) + ON CONFLICT(room_id) DO UPDATE SET schema_version=excluded.schema_version""", + (room_id, _TRANSCRIPT_SCHEMA_VERSION), + ) + return cursor + def sync(self, *, room_id: str, latest_seq: int) -> int: """Materialize each unseen event exactly once by durable cursor.""" - with self._connect() as conn: conn.execute("BEGIN IMMEDIATE") - _require_room(conn, room_id) - conn.execute( - """INSERT OR IGNORE INTO hosted_room_policy_cursors( - room_id, through_seq, stopped_through_seq, updated_at - ) VALUES (?, 0, 0, 0)""", - (room_id,), - ) - row = conn.execute( - "SELECT through_seq FROM hosted_room_policy_cursors WHERE room_id=?", - (room_id,), - ).fetchone() - cursor = int(row["through_seq"]) - transcript_state = conn.execute( - """SELECT schema_version - FROM hosted_room_policy_transcript_state WHERE room_id=?""", - (room_id,), - ).fetchone() - if ( - transcript_state is None - or int(transcript_state["schema_version"]) - < _TRANSCRIPT_SCHEMA_VERSION - ): - self._backfill_transcript( - conn, - room_id=room_id, - through_seq=cursor, - ) - conn.execute( - """INSERT INTO hosted_room_policy_transcript_state( - room_id, schema_version - ) VALUES (?, ?) - ON CONFLICT(room_id) DO UPDATE SET - schema_version=excluded.schema_version""", - (room_id, _TRANSCRIPT_SCHEMA_VERSION), - ) + cursor = self._ensure_cursor_and_transcript(conn, room_id) if cursor > latest_seq: raise RuntimeError("room policy cursor is ahead of the durable log") while cursor < latest_seq: page = hosted_rooms.read_events( - self.db_path, - room_id=room_id, - since_seq=cursor, - limit=hosted_rooms.MAX_LOG_LIMIT, + self.db_path, room_id=room_id, since_seq=cursor, limit=hosted_rooms.MAX_LOG_LIMIT ) - rows = [ - event for event in page.get("events", []) if isinstance(event, Mapping) - ] + rows = [event for event in page.get("events", []) if isinstance(event, Mapping)] next_cursor = int(page.get("cursor") or cursor) if not rows or next_cursor <= cursor: raise RuntimeError("hosted room policy cursor did not advance") @@ -535,13 +372,8 @@ class HostedRoomPolicyCheckpoint: for event in rows: self._apply_event(conn, event) updated = conn.execute( - """UPDATE hosted_room_policy_cursors - SET through_seq=?, updated_at=? WHERE room_id=?""", - ( - next_cursor, - float(rows[-1].get("created_at") or 0), - room_id, - ), + "UPDATE hosted_room_policy_cursors SET through_seq=?, updated_at=? WHERE room_id=?", + (next_cursor, float(rows[-1].get("created_at") or 0), room_id), ) if updated.rowcount != 1: raise RuntimeError("room policy cursor disappeared during replay") @@ -550,47 +382,37 @@ class HostedRoomPolicyCheckpoint: def snapshot(self, *, room_id: str, latest_seq: int) -> PolicySnapshot: """Return only the oldest active discussion and its watermark set.""" - through_seq = self.sync(room_id=room_id, latest_seq=latest_seq) with self._connect() as conn: cursor = conn.execute( - """SELECT stopped_through_seq FROM hosted_room_policy_cursors - WHERE room_id=?""", + "SELECT stopped_through_seq FROM hosted_room_policy_cursors WHERE room_id=?", (room_id,), ).fetchone() stopped_through_seq = int(cursor["stopped_through_seq"]) thread = conn.execute( - """SELECT thread_id, discussion_event_id - FROM hosted_room_policy_threads + """SELECT thread_id, discussion_event_id FROM hosted_room_policy_threads WHERE room_id=? AND completed=0 AND latest_user_seq>? ORDER BY latest_user_seq, thread_id LIMIT 1""", (room_id, stopped_through_seq), ).fetchone() if thread is None: return PolicySnapshot( - through_seq=through_seq, - stopped_through_seq=stopped_through_seq, - events=(), - watermarks={}, + through_seq=through_seq, stopped_through_seq=stopped_through_seq, + events=(), watermarks={}, ) thread_id = str(thread["thread_id"]) events = self._discussion_events( - conn, - room_id=room_id, - thread_id=thread_id, + conn, room_id=room_id, thread_id=thread_id, discussion_event_id=str(thread["discussion_event_id"]), bound_error="active room policy projection exceeded its bound", ) watermark_rows = conn.execute( - """SELECT member_id, seen_through_seq - FROM hosted_room_policy_watermarks + """SELECT member_id, seen_through_seq FROM hosted_room_policy_watermarks WHERE room_id=? AND thread_id=?""", (room_id, thread_id), ).fetchall() return PolicySnapshot( - through_seq=through_seq, - stopped_through_seq=stopped_through_seq, - events=tuple(events), + through_seq=through_seq, stopped_through_seq=stopped_through_seq, events=tuple(events), watermarks={ (thread_id, str(row["member_id"])): int(row["seen_through_seq"]) for row in watermark_rows @@ -598,77 +420,44 @@ class HostedRoomPolicyCheckpoint: ) def publication_exists( - self, - *, - room_id: str, - task_id: str, - status: str, - execution_generation: int, + self, *, room_id: str, task_id: str, status: str, execution_generation: int ) -> bool: """Return whether one exact driver outcome is already in the room log.""" - - kind = f"turn.{status}" - generation = execution_generation if status == "deferred" else 0 + if status == "deferred": + sql = """SELECT 1 FROM hosted_room_policy_publications + WHERE room_id=? AND task_id=? AND kind=? AND execution_generation=?""" + params = (room_id, task_id, f"turn.{status}", execution_generation) + else: + sql = """SELECT 1 FROM hosted_room_policy_publications + WHERE room_id=? AND task_id=? AND kind IN ('turn.settled', 'turn.failed', 'turn.cancelled')""" + params = (room_id, task_id) with self._connect() as conn: - if status == "deferred": - row = conn.execute( - """SELECT 1 FROM hosted_room_policy_publications - WHERE room_id=? AND task_id=? AND kind=? - AND execution_generation=?""", - (room_id, task_id, kind, generation), - ).fetchone() - else: - row = conn.execute( - """SELECT 1 FROM hosted_room_policy_publications - WHERE room_id=? AND task_id=? AND kind IN ( - 'turn.settled', 'turn.failed', 'turn.cancelled' - )""", - (room_id, task_id), - ).fetchone() - return row is not None + return conn.execute(sql, params).fetchone() is not None - def events_for_task( - self, - *, - room_id: str, - source_event_seq: int, - ) -> list[dict[str, Any]]: + def events_for_task(self, *, room_id: str, source_event_seq: int) -> list[dict[str, Any]]: """Load one bounded discussion projection for terminal reconstruction.""" - with self._connect() as conn: source = conn.execute( - """SELECT discussion_event_id, thread_id - FROM hosted_room_policy_events - WHERE room_id=? AND seq=?""", + "SELECT discussion_event_id, thread_id FROM hosted_room_policy_events WHERE room_id=? AND seq=?", (room_id, source_event_seq), ).fetchone() if source is None: return [] return self._discussion_events( - conn, - room_id=room_id, - thread_id=str(source["thread_id"]), + conn, room_id=room_id, thread_id=str(source["thread_id"]), discussion_event_id=str(source["discussion_event_id"]), bound_error="task policy projection exceeded its bound", ) def compact_completed(self, *, room_id: str) -> None: """Drop any completed projections left by an interrupted sync.""" - with self._connect() as conn: completed = conn.execute( - """SELECT discussion_event_id FROM hosted_room_policy_threads - WHERE room_id=? AND completed=1""", + "SELECT discussion_event_id FROM hosted_room_policy_threads WHERE room_id=? AND completed=1", (room_id,), ).fetchall() for row in completed: - conn.execute( - """DELETE FROM hosted_room_policy_events - WHERE room_id=? AND discussion_event_id=?""", - (room_id, str(row["discussion_event_id"])), - ) + conn.execute(_DELETE_ACTIVE_EVENTS_SQL, (room_id, str(row["discussion_event_id"]))) conn.execute( - """DELETE FROM hosted_room_policy_threads - WHERE room_id=? AND completed=1""", - (room_id,), + "DELETE FROM hosted_room_policy_threads WHERE room_id=? AND completed=1", (room_id,) )