diff --git a/gateway/hosted_room_discussion.py b/gateway/hosted_room_discussion.py index af376853a9..0dc4e9b225 100644 --- a/gateway/hosted_room_discussion.py +++ b/gateway/hosted_room_discussion.py @@ -1,10 +1,9 @@ """Deterministic policy for same-gateway hosted-room Discussions. -Pure (no I/O, transport or model knowledge): a frozen local roster plus the complete -typed room log yields at most one next driver task. Discussion coordinates live in -deterministic ``TaskIdentity`` values and typed terminal events rather than a widened -driver payload, so a restart reconstructs tasks from durable state. Callers must -reconcile terminal driver rows into publication plans before asking for the next task. +Pure (no I/O, transport or model knowledge): a frozen local roster plus the complete typed room log yields +at most one next driver task. Discussion coordinates live in deterministic ``TaskIdentity`` values and typed +terminal events rather than a widened driver payload, so a restart reconstructs tasks from durable state. +Callers must reconcile terminal driver rows into publication plans before asking for the next task. """ from __future__ import annotations @@ -78,7 +77,6 @@ class DiscussionReconstructionError(DiscussionPolicyError): @dataclass(frozen=True) class DiscussionMember: """One immutable local or peer member of the hosted room.""" - member_id: str profile: str handle: str @@ -89,7 +87,6 @@ class DiscussionMember: @dataclass(frozen=True) class DiscussionRoom: """Validated policy projection of one active hosted room.""" - room_id: str name: str members: tuple[DiscussionMember, ...] @@ -100,7 +97,6 @@ class DiscussionRoom: @dataclass(frozen=True) class DiscussionTaskPlan: """One deterministic member turn compatible with the driver schema.""" - identity: driver.TaskIdentity payload: Mapping[str, Any] discussion_event_id: str @@ -113,7 +109,6 @@ class DiscussionTaskPlan: @dataclass(frozen=True) class DiscussionDecision: """Current result of replaying one room's Discussion policy.""" - status: DecisionStatus reason: str discussion_event_id: str | None = None @@ -125,7 +120,6 @@ class DiscussionDecision: @dataclass(frozen=True) class EventPlan: """One idempotent append for :func:`gateway.hosted_rooms.append_event`.""" - event_id: str kind: str actor: Mapping[str, str] @@ -141,7 +135,6 @@ class EventPlan: @dataclass(frozen=True) class PublicationPlan: """Ordered visible and terminal effects for one driver task.""" - task_id: str terminal_kind: str events: tuple[EventPlan, ...] @@ -160,6 +153,7 @@ class _ValidatedEvent: _identifier = partial(common.identifier, error=DiscussionValidationError, max_chars=driver.MAX_IDENTIFIER_CHARS) _exact_fields = partial(common.exact_fields, error=DiscussionValidationError) _bounded_int = partial(common.bounded_int, error=DiscussionValidationError) +_text = partial(common.text, error=DiscussionValidationError) def _positive_int(value: Any, *, label: str) -> int: @@ -170,9 +164,6 @@ def _zero_based_int(value: Any, *, label: str, maximum: int) -> int: 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_id(member: DiscussionMember) -> str | None: target = member.target return target.get("peer_id") if target and target.get("kind") == "peer" else None @@ -196,8 +187,9 @@ def _all_failure_reasons() -> frozenset[str]: def validate_user_payload(value: Any) -> dict[str, Any]: """Validate and normalize the exact ``message.user`` Discussion payload.""" payload = _exact_fields(value, label="user payload", required=_USER_PAYLOAD_FIELDS) - text = _text(payload["text"], label="user payload text", max_bytes=MAX_USER_TEXT_BYTES) - return {"text": text, "thread_id": _identifier(payload["thread_id"], label="thread_id")} + return { + "text": _text(payload["text"], label="user payload text", max_bytes=MAX_USER_TEXT_BYTES), + "thread_id": _identifier(payload["thread_id"], label="thread_id")} def _validate_member_target(value: Any, *, profile: str, known_profiles: set[str], index: int) -> dict[str, Any]: @@ -218,13 +210,13 @@ def _validate_member_target(value: Any, *, profile: str, known_profiles: set[str return {"kind": "local", "profile": profile} if target_profile != profile: raise DiscussionValidationError(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): + digest = target["capability_digest"] + if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", 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"), - "profile": target_profile, "capability_digest": capability_digest} + "profile": target_profile, "capability_digest": digest} def _validate_member(raw: Any, index: int, known_profiles: set[str]) -> DiscussionMember: @@ -240,8 +232,7 @@ def _validate_member(raw: Any, index: int, known_profiles: set[str]) -> Discussi _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): + if not isinstance(display_name := member.get("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") @@ -281,8 +272,7 @@ def validate_room(value: Any, *, local_profiles: Iterable[str]) -> DiscussionRoo if value.get("disbanded_at") is not None: raise DiscussionValidationError("room is disbanded") room_id = _identifier(value.get("room_id"), label="room_id") - name = value.get("name") - if not isinstance(name, str) or not name.strip(): + if not isinstance(name := value.get("name"), str) or not name.strip(): raise DiscussionValidationError("room name must be a non-empty string") if len(name := name.strip()) > hosted_rooms.MAX_ROOM_NAME_CHARS: raise DiscussionValidationError("room name is too long") @@ -327,14 +317,12 @@ def _unaddressed_member_mentions( continue speaker_id = str(event.payload["member_id"]) last_post_at[speaker_id] = event.seq - cited = resolve_mentions((str(event.payload["text"]),), room.members, default_all=False) - for member in cited: + for member in resolve_mentions((str(event.payload["text"]),), room.members, default_all=False): if member.member_id != speaker_id: cited_at[member.member_id] = event.seq return tuple( 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]) + if member.member_id in cited_at and last_post_at.get(member.member_id, 0) <= cited_at[member.member_id]) def _require_gateway_actor(actor: Mapping[str, Any], room: DiscussionRoom, message: str) -> None: @@ -359,8 +347,6 @@ def _validate_turn_coordinates(payload: Mapping[str, Any], room: DiscussionRoom) # -- 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: Payload, actor: Payload, room: DiscussionRoom) -> Payload: payload = validate_user_payload(payload) if actor.get("kind") != "user": @@ -371,8 +357,7 @@ def _validate_user_event(kind: str, payload: Payload, actor: Payload, room: Disc 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): + if not isinstance(text := payload.get("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")) expected = {**_member_actor(member, display_name=False), "connection_id": _peer_id(member)} @@ -475,8 +460,7 @@ def _derive_member_watermarks(events: Sequence[_ValidatedEvent]) -> dict[tuple[s if event.kind not in _TERMINAL_EVENT_KINDS: continue task_id = str(event.payload["task_id"]) - previous = terminal_by_task.get(task_id) - if previous is not None: + if (previous := terminal_by_task.get(task_id)) is not None: if previous.kind != "turn.deferred": raise DiscussionValidationError(f"task '{task_id}' has more than one terminal room event") if event.kind == "turn.deferred" and int( @@ -488,8 +472,7 @@ def _derive_member_watermarks(events: Sequence[_ValidatedEvent]) -> dict[tuple[s if event.kind == "turn.settled" and not event.payload["passed"]: message = messages_by_id.get(str(event.payload["message_event_id"])) if message is None or any( - message.payload.get(field) != event.payload.get(field) - for field in ("task_id", "member_id", "thread_id")): + message.payload.get(f) != event.payload.get(f) for f in ("task_id", "member_id", "thread_id")): raise DiscussionValidationError("turn.settled references no matching member message") watermark = max(watermark, message.seq) watermarks[key] = max(watermarks.get(key, 0), watermark) @@ -539,20 +522,16 @@ def _build_prompt( fixed_bytes = len("\n".join([*opening, *rules]).encode("utf-8")) available = max(0, driver.MAX_PROMPT_BYTES - fixed_bytes - 1) selected: list[str] = [] - omitted = False for event in reversed(delta): line = f" {_format_message(event, room)}" - if (line_bytes := len(line.encode("utf-8")) + 1) <= available: - selected.append(line) - available -= line_bytes - continue - if not selected and available > 32: - selected.append(_truncate_utf8_text(line, max_bytes=available)) - omitted = True - break + if (line_bytes := len(line.encode("utf-8")) + 1) > available: + if not selected and available > 32: + selected.append(_truncate_utf8_text(line, max_bytes=available)) + selected.append(" [Earlier content omitted to fit this turn.]") + break + selected.append(line) + available -= line_bytes selected.reverse() - if omitted: - selected.insert(0, " [Earlier content omitted to fit this turn.]") 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,8 +560,7 @@ def _pending_discussion(validated: Sequence[_ValidatedEvent]) -> _ValidatedEvent """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) completed_discussion_ids = { - str(event.payload["discussion_event_id"]) - for event in validated + 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 = { str(event.payload["thread_id"]): event for event in validated if event.kind == "message.user"} @@ -597,26 +575,20 @@ def _thread_messages( """Return (thread messages, messages since the discussion, this discussion's member messages).""" thread_id = str(discussion.payload["thread_id"]) committed_member_message_ids = { - str(event.payload["message_event_id"]) - for event in validated + 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} # 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 # policy input yet: ignoring it reproduces the original task coordinates so # the caller can inspect the terminal driver row and finish publication. thread_messages = tuple( - event - for event in validated - if event.payload.get("thread_id") == thread_id - and ( + event for event in validated + if event.payload.get("thread_id") == thread_id and ( event.kind == "message.user" 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) - member_messages = tuple( + return thread_messages, tuple(event for event in thread_messages if event.seq >= discussion.seq), tuple( 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 + if event.kind == "message.member" and event.payload.get("discussion_event_id") == discussion.event_id) def _effective_watermarks( @@ -636,8 +608,7 @@ def plan_next_task( """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: + if (discussion := _pending_discussion(validated)) is None: return DiscussionDecision(status="idle", reason="no_pending_user_event") thread_id = str(discussion.payload["thread_id"]) decide = partial( @@ -647,10 +618,8 @@ def plan_next_task( if len(member_messages) >= MAX_DISCUSSION_MESSAGES: return decide("bounded", "max_messages") terminals = { - (int(event.payload["round_index"]), str(event.payload["member_id"])): event - for event in validated - if event.kind in _TERMINAL_EVENT_KINDS - and event.payload.get("discussion_event_id") == discussion.event_id} + (int(event.payload["round_index"]), str(event.payload["member_id"])) for event in validated + if event.kind in _TERMINAL_EVENT_KINDS and event.payload.get("discussion_event_id") == discussion.event_id} 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): @@ -660,8 +629,7 @@ def plan_next_task( # watermark remains intact, so a peer cited later still receives the # complete bounded transcript delta without consuming turns meanwhile. responders = ( - resolve_mentions((str(discussion.payload["text"]),), room.members) - if round_index == 0 + resolve_mentions((str(discussion.payload["text"]),), room.members) if round_index == 0 else _unaddressed_member_mentions(discussion_messages, room)) for member_index, member in enumerate(_rotate(responders, round_index)): if (round_index, member.member_id) in terminals: @@ -672,10 +640,9 @@ def plan_next_task( prompt = _build_prompt( room=room, member=member, messages=thread_messages, watermark=watermark, seen_through_seq=seen_through_seq) - task = _make_task_plan( + return decide("task", "member_turn", 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) - return decide("task", "member_turn", task=task) + round_index=round_index, seen_through_seq=seen_through_seq, prompt=prompt)) 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: @@ -693,8 +660,8 @@ def reconstruct_task_plan( if not isinstance(identity, driver.TaskIdentity) or not isinstance(payload, Mapping): raise DiscussionReconstructionError("driver task has no valid identity or payload") fields = frozenset(payload) - if not driver._TASK_PAYLOAD_REQUIRED_FIELDS <= fields or ( - fields - driver._TASK_PAYLOAD_REQUIRED_FIELDS - driver._TASK_PAYLOAD_OPTIONAL_FIELDS): + if not driver._TASK_PAYLOAD_REQUIRED_FIELDS <= fields <= ( + driver._TASK_PAYLOAD_REQUIRED_FIELDS | driver._TASK_PAYLOAD_OPTIONAL_FIELDS): raise DiscussionReconstructionError("driver task payload shape changed") if (match := _TURN_ID_RE.fullmatch(identity.turn_id)) is None: raise DiscussionReconstructionError("turn_id is not a Discussion coordinate") @@ -708,13 +675,11 @@ def reconstruct_task_plan( raise DiscussionReconstructionError("task identity does not match its room thread") 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) + m for m in room.members + if m.profile == profile and (target_member_id is None or m.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") - if not isinstance(prompt, str) or not prompt.strip(): + if not isinstance(prompt := payload.get("prompt"), str) or not prompt.strip(): raise DiscussionReconstructionError("task prompt is missing") if len(prompt.encode("utf-8")) > driver.MAX_PROMPT_BYTES: raise DiscussionReconstructionError("task prompt exceeds the driver limit") @@ -742,11 +707,11 @@ def _terminal_text(result: Any, *, field: str, fallback: str) -> str: # -- per-status terminal payload builders (dispatched by plan_publication) ----- # Each returns (extra terminal payload fields, visible effects to publish first). +Effects = tuple[dict[str, Any], list[EventPlan]] def _settled_effects( - result: Any, *, task: DiscussionTaskPlan, room: DiscussionRoom, message_event_id: str, **_: Any -) -> tuple[dict[str, Any], list[EventPlan]]: + result: Any, *, task: DiscussionTaskPlan, room: DiscussionRoom, message_event_id: str, **_: Any) -> Effects: text = _truncate_utf8_text( _terminal_text(result, field="text", fallback=""), max_bytes=MAX_MEMBER_TEXT_BYTES, suffix=_TRUNCATED_REPLY_NOTICE) @@ -758,7 +723,7 @@ def _settled_effects( authority_epoch=room.authority_epoch)] -def _failed_effects(result: Any, **_: Any) -> tuple[dict[str, Any], list[EventPlan]]: +def _failed_effects(result: Any, **_: Any) -> Effects: error_text = _terminal_text(result, field="error", fallback="member turn failed") from tools.bot_failure_reasons import classify_agent_error supplied_reason = ( @@ -767,17 +732,15 @@ def _failed_effects(result: Any, **_: Any) -> tuple[dict[str, Any], list[EventPl return {"error": error_text, "reason_code": reason_code}, [] -def _cancelled_effects(result: Any, *, newer_same_thread: bool, **_: Any) -> tuple[dict[str, Any], list[EventPlan]]: +def _cancelled_effects(result: Any, *, newer_same_thread: bool, **_: Any) -> Effects: 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( - 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")}, [] +def _deferred_effects(result: Any, *, execution_generation: int | None, **_: Any) -> Effects: + reason = _terminal_text(result, field="reason", fallback="member_unavailable") + return {"execution_generation": execution_generation, "reason": reason}, [] _TERMINAL_EFFECTS = { @@ -790,25 +753,22 @@ def plan_publication( result: Any = None, execution_generation: int | None = None, local_profiles: Iterable[str]) -> PublicationPlan: """Plan idempotent room effects for one terminal driver task. - A newer user event in the same thread supersedes a late result: the task - stays terminal in driver state, but only a deterministic cancellation is - published so stale prose and its watermark cannot hide the newer message. + A newer user event in the same thread supersedes a late result: the task stays terminal in driver state, + but only a deterministic cancellation is published so stale prose and its watermark cannot hide it. """ room = validate_room(room_value, local_profiles=local_profiles) validated = _validated_events(events, room=room) - if task.identity.room_id != room.room_id: - 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 _TERMINAL_EFFECTS: - raise DiscussionValidationError("invalid terminal publication status") + for failed, message in ( + (task.identity.room_id != room.room_id, "task belongs to a different room"), + (task.member not in room.members, "task member is not in the frozen roster"), + (status not in _TERMINAL_EFFECTS, "invalid terminal publication status")): + if failed: + raise DiscussionValidationError(message) if status == "deferred": _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 - and event.payload.get("thread_id") == task.identity.thread_id - for event in validated) + event.kind == "message.user" and event.seq > task.seen_through_seq + and event.payload.get("thread_id") == task.identity.thread_id for event in validated) effective_status: TerminalKind = ("cancelled" if newer_same_thread and status != "deferred" else status) digest = task.identity.task_id.removeprefix("dtask:") terminal_event_id = ( diff --git a/gateway/hosted_room_driver.py b/gateway/hosted_room_driver.py index ac902cf7f6..fb7268767a 100644 --- a/gateway/hosted_room_driver.py +++ b/gateway/hosted_room_driver.py @@ -1,9 +1,8 @@ """Durable execution state for a same-gateway hosted room driver. -Owns only the driver lease and task state machine: no model calls, no -sessions, no dependency on the hosted-room event log. Callers supply the -database path and clock so recovery and fencing are testable without -process-global state. +Owns only the driver lease and task state machine: no model calls, no sessions, no dependency on the +hosted-room event log. Callers supply the database path and clock so recovery and fencing are testable +without process-global state. """ from __future__ import annotations @@ -19,7 +18,8 @@ from functools import partial from typing import Any, Callable, Literal, get_args from gateway.hosted_rooms_common import ( - DbPath, bounded_int, canonical_json, compact_json, connect, identifier, table_columns, text, transaction) + DbPath, bounded_int, canonical_json, compact_json, connect, fenced_update, identifier, table_columns, text, + transaction) Clock = Callable[[], float] TaskStatus = Literal["queued", "running", "settled", "failed", "cancelled", "indeterminate", "deferred", "stopping"] @@ -52,8 +52,8 @@ _TASK_INDEX_SQL = """CREATE INDEX {if_not_exists}idx_hosted_room_driver_tasks_st ON hosted_room_driver_tasks(room_id, status, source_event_seq, created_at, task_id)""" # --- Fenced task UPDATE statements (one per state-machine transition) --------- -# Every transition is "UPDATE ... SET WHERE room_id=? AND task_id=? AND "; -# the fence names the expected status plus the generations that must not have moved. +# Every transition is "UPDATE ... SET WHERE room_id=? AND task_id=? AND "; the fence names the +# expected status plus the generations that must not have moved. _GENERATION_FENCE = "execution_generation=? AND cancel_generation=?" _RUN_FENCE = "run_gateway_id=? AND run_process_generation=? AND run_lease_generation=?" _SETTLE_SET = "status=?, settlement_id=?, settlement_status=?, result_json=?, terminal_at=?, updated_at=?" @@ -133,15 +133,13 @@ def _timestamp(clock: Clock) -> float: return _finite(clock, "clock must return a finite number") -def _ttl(value: Any) -> float: - return _finite(lambda: value, "ttl_seconds must be a finite positive number", positive=True) - - -def _expiry(now: float, ttl: float) -> float: - expires_at = now + ttl - if not math.isfinite(expires_at): +def _lease_window(ttl_seconds: Any, clock: Clock) -> tuple[float, float]: + """Validate ttl (first) and clock -> ``(now, expires_at)``; the sum itself must stay finite.""" + ttl = _finite(lambda: ttl_seconds, "ttl_seconds must be a finite positive number", positive=True) + now = _timestamp(clock) + if not math.isfinite(now + ttl): raise DriverValidationError("lease expiry must be finite") - return expires_at + return now, now + ttl def _task_payload(value: Any) -> tuple[dict[str, Any], str, str]: @@ -167,7 +165,6 @@ def _task_payload(value: Any) -> tuple[dict[str, Any], str, str]: @dataclass(frozen=True) class TaskIdentity: """Stable identity for one admitted room turn.""" - room_id: str task_id: str thread_id: str @@ -181,7 +178,6 @@ class TaskIdentity: @dataclass(frozen=True) class DriverLease: """A fenced lease held by one gateway process incarnation.""" - room_id: str gateway_id: str authority_epoch: int @@ -194,7 +190,6 @@ class DriverLease: @dataclass(frozen=True) class TaskAttempt: """The exact running generation authorized to settle one task.""" - identity: TaskIdentity lease: DriverLease execution_generation: int @@ -202,8 +197,6 @@ class TaskAttempt: def _create_task_table(conn: sqlite3.Connection, table: str = "hosted_room_driver_tasks") -> None: - if table not in {"hosted_room_driver_tasks", "hosted_room_driver_tasks_next"}: - raise DriverStateError("invalid hosted-room task table name") conn.execute( f"""CREATE TABLE IF NOT EXISTS {table} ( room_id TEXT NOT NULL, task_id TEXT NOT NULL, thread_id TEXT NOT NULL, turn_id TEXT NOT NULL, @@ -233,15 +226,15 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: def _validate_schema(conn: sqlite3.Connection) -> None: - lease_columns = table_columns(conn, "hosted_room_driver_leases") - task_columns = table_columns(conn, "hosted_room_driver_tasks") - if lease_columns != _LEASE_COLUMNS or task_columns != _TASK_COLUMNS: + columns = table_columns(conn, "hosted_room_driver_leases"), table_columns(conn, "hosted_room_driver_tasks") + if columns != (_LEASE_COLUMNS, _TASK_COLUMNS): raise DriverStateError( "unsupported unpublished hosted-room driver schema; " "recreate the driver tables before starting the driver") for table in ("hosted_room_driver_leases", "hosted_room_driver_tasks"): - foreign_keys = conn.execute(f"PRAGMA foreign_key_list({table})").fetchall() - if not any(row[2] == "hosted_rooms" and row[3] == "room_id" and row[4] == "room_id" for row in foreign_keys): + if not any( + row[2] == "hosted_rooms" and row[3] == "room_id" and row[4] == "room_id" + for row in conn.execute(f"PRAGMA foreign_key_list({table})").fetchall()): raise DriverStateError(f"{table} is missing its hosted_rooms foreign key") @@ -255,13 +248,6 @@ def _schema_objects_exist(conn: sqlite3.Connection) -> bool: return index is not None -def _task_schema_supports_current_statuses(conn: sqlite3.Connection) -> bool: - row = conn.execute( - "SELECT sql FROM sqlite_master WHERE type='table' AND name='hosted_room_driver_tasks'").fetchone() - sql = str(row[0] or "").lower() if row else "" - return "'stopping'" in sql and "'deferred'" in sql - - def _migrate_task_status_constraint(conn: sqlite3.Connection) -> None: """Expand the unpublished task-state CHECK without losing durable work.""" conn.execute("DROP INDEX IF EXISTS idx_hosted_room_driver_tasks_status") @@ -276,13 +262,16 @@ def _migrate_task_status_constraint(conn: sqlite3.Connection) -> None: def _connect(db_path: DbPath) -> sqlite3.Connection: """Open the store; existing tables are validated (after the status-constraint migration when needed). - - The driver schema never shipped, so an incompatible draft fails closed in ``_validate_schema``. - """ + The driver schema never shipped, so an incompatible draft fails closed in ``_validate_schema``.""" existing: list[bool] = [] def ready(conn: sqlite3.Connection) -> bool: existing.append(_schema_objects_exist(conn)) - return existing[0] and _task_schema_supports_current_statuses(conn) + if not existing[0]: + return False + row = conn.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name='hosted_room_driver_tasks'").fetchone() + sql = str(row[0] or "").lower() if row else "" + return "'stopping'" in sql and "'deferred'" in sql # task-status CHECK already covers the current states conn = connect( db_path, db_label="state.db (hosted_room_driver)", ready=ready, initialize=lambda conn: (_migrate_task_status_constraint if existing[0] else _initialize_schema)(conn)) @@ -338,10 +327,12 @@ def _task_from_row(row: sqlite3.Row, *, idempotent: bool = False) -> dict[str, A return task -def _load_task(conn: sqlite3.Connection, identity: TaskIdentity) -> sqlite3.Row: +def _load_task(conn: sqlite3.Connection, identity: TaskIdentity, *, required: bool = True) -> sqlite3.Row | None: row = conn.execute(_SELECT_TASK, (identity.room_id, identity.task_id)).fetchone() if row is None: - raise TaskConflictError("task does not exist") + if required: + raise TaskConflictError("task does not exist") + return None if _task_identity_from_row(row) != identity: raise TaskConflictError("task_id is already bound to a different turn") return row @@ -362,10 +353,8 @@ def _load_active_room(conn: sqlite3.Connection, room_id: str) -> sqlite3.Row: if "no such table" in str(exc).lower(): raise RoomUnavailableError("hosted room does not exist") from exc raise - if row is None: - raise RoomUnavailableError("hosted room does not exist") - if row["disbanded_at"] is not None: - raise RoomUnavailableError("hosted room is disbanded") + if row is None or row["disbanded_at"] is not None: + raise RoomUnavailableError("hosted room does not exist" if row is None else "hosted room is disbanded") return row @@ -376,8 +365,9 @@ def _require_room_authority(conn: sqlite3.Connection, room_id: str, gateway_id: return room -def _require_lease_authority(conn: sqlite3.Connection, lease: DriverLease) -> sqlite3.Row: - return _require_room_authority(conn, lease.room_id, lease.gateway_id, lease.authority_epoch) +def _run_fence(lease: DriverLease) -> tuple[str, str, int]: + """The ``_RUN_FENCE`` bind order: (gateway_id, process_generation, lease_generation).""" + return lease.gateway_id, lease.process_generation, lease.lease_generation def _lease_row_matches(row: sqlite3.Row | None, lease: DriverLease) -> bool: @@ -387,7 +377,7 @@ def _lease_row_matches(row: sqlite3.Row | None, lease: DriverLease) -> bool: def _require_active_lease(conn: sqlite3.Connection, lease: DriverLease, *, now: float) -> sqlite3.Row: - _require_lease_authority(conn, lease) + _require_room_authority(conn, lease.room_id, lease.gateway_id, lease.authority_epoch) row = conn.execute(_SELECT_LEASE, (lease.room_id,)).fetchone() if not _lease_row_matches(row, lease) or row["released_at"] is not None or float(row["expires_at"]) <= now: raise StaleLeaseError("driver lease is stale or expired") @@ -424,10 +414,8 @@ def _terminal_settlement_id(settlement_id: Any, status: Any) -> str: def _settlement( settlement_id: Any, status: Any, result: Any, clock: Clock ) -> tuple[float, Callable[[sqlite3.Row], Any], tuple[Any, ...]]: - """Validate one terminal settlement -> (now, replay predicate, ``_SETTLE_SET`` params). - - Replay: an identical settlement already committed is idempotent; a different one is a conflict. - """ + """Validate one terminal settlement -> (now, replay predicate, ``_SETTLE_SET`` params); the replay treats an + identical committed settlement as idempotent and a different one as a conflict.""" settlement_id = _terminal_settlement_id(settlement_id, status) result_json = _canonical_json(result) now = _timestamp(clock) @@ -463,9 +451,9 @@ def _transition( guard: Callable[[sqlite3.Row], None] | None = None) -> dict[str, Any]: """Run one fenced task transition: load -> idempotent replay -> lease/fence guard -> UPDATE. - ``sql`` binds ``(*set_params, room_id, task_id, *fence_params)`` and must hit exactly one row or - ``stale`` is raised. ``lease_first`` checks the lease before the row load (recovery paths) instead - of after the replay (settlement paths: an identical replay still succeeds after the lease moved on). + ``sql`` binds ``(*set_params, room_id, task_id, *fence_params)`` and must hit exactly one row or ``stale`` + is raised. ``lease_first`` checks the lease before the row load (recovery paths) instead of after the + replay (settlement paths: an identical replay still succeeds after the lease moved on). """ params = (*set_params, identity.room_id, identity.task_id, *fence_params) with _transaction(db_path) as conn: @@ -478,8 +466,7 @@ def _transition( _require_active_lease(conn, lease, now=now) if guard is not None: guard(row) - if conn.execute(sql, params).rowcount != 1: - raise StaleTaskError(stale) + fenced_update(conn, sql, params, StaleTaskError(stale)) return _task_from_row(_load_task(conn, identity)) @@ -501,22 +488,16 @@ def _run_fence_transition( db_path: DbPath, attempt: TaskAttempt, *, guard_stale: str, lease_generation: Callable[[Any], int] = int, **transition: Any) -> dict[str, Any]: """Transition fenced on this attempt's running generation under its exact lease (row guard + SQL fence). - - ``lease_generation`` casts the stored run_lease_generation: ``int`` raises on NULL, ``int(v or 0)`` reads 0. - """ + ``lease_generation`` casts the stored run_lease_generation: ``int`` raises on NULL, ``int(v or 0)`` reads 0.""" lease = attempt.lease def guard(row: sqlite3.Row) -> None: - if not ( - _generations_match(row, "running", attempt.execution_generation, attempt.cancel_generation) - and row["run_gateway_id"] == lease.gateway_id - and row["run_process_generation"] == lease.process_generation - and lease_generation(row["run_lease_generation"]) == lease.lease_generation): + if not _generations_match(row, "running", attempt.execution_generation, attempt.cancel_generation) or ( + row["run_gateway_id"], row["run_process_generation"], lease_generation(row["run_lease_generation"]) + ) != _run_fence(lease): raise StaleTaskError(guard_stale) return _transition( db_path, attempt.identity, lease=lease, guard=guard, - fence_params=( - attempt.execution_generation, attempt.cancel_generation, lease.gateway_id, lease.process_generation, - lease.lease_generation), **transition) + fence_params=(attempt.execution_generation, attempt.cancel_generation, *_run_fence(lease)), **transition) def acquire_lease( @@ -527,9 +508,7 @@ def acquire_lease( gateway_id = _identifier(gateway_id, label="gateway_id") authority_epoch = _bounded_int(authority_epoch, message="authority_epoch must be a positive integer", low=1) process_generation = _identifier(process_generation, label="process_generation") - ttl_seconds = _ttl(ttl_seconds) - now = _timestamp(clock) - expires_at = _expiry(now, ttl_seconds) + now, expires_at = _lease_window(ttl_seconds, clock) with _transaction(db_path) as conn: _require_room_authority(conn, room_id, gateway_id, authority_epoch) row = conn.execute(_SELECT_LEASE, (room_id,)).fetchone() @@ -550,33 +529,29 @@ def acquire_lease( return _lease_from_row({**dict(row), "expires_at": renewed_expiry}) if same_authority and live: raise LeaseHeldError("room driver lease is held by another generation") - updated = conn.execute("""UPDATE hosted_room_driver_leases + fenced_update(conn, """UPDATE hosted_room_driver_leases SET gateway_id=?, authority_epoch=?, process_generation=?, lease_generation=lease_generation + 1, expires_at=?, acquired_at=?, updated_at=?, released_at=NULL WHERE room_id=? AND lease_generation=? AND ( gateway_id != ? OR authority_epoch != ? OR released_at IS NOT NULL OR expires_at <= ?)""", ( gateway_id, authority_epoch, process_generation, expires_at, now, now, room_id, - int(row["lease_generation"]), gateway_id, authority_epoch, now)) - if updated.rowcount != 1: - raise LeaseHeldError("room driver lease changed during acquisition") + int(row["lease_generation"]), gateway_id, authority_epoch, now), + LeaseHeldError("room driver lease changed during acquisition")) return _lease_from_row(conn.execute(_SELECT_LEASE, (room_id,)).fetchone(), reclaimed=True) def renew_lease(db_path: DbPath, lease: DriverLease, *, ttl_seconds: Any, clock: Clock) -> DriverLease: """Renew the exact active lease generation or fail closed.""" - ttl_seconds = _ttl(ttl_seconds) - now = _timestamp(clock) - requested_expiry = _expiry(now, ttl_seconds) + now, requested_expiry = _lease_window(ttl_seconds, clock) with _transaction(db_path) as conn: current = _require_active_lease(conn, lease, now=now) expires_at = max(float(current["expires_at"]), requested_expiry) - updated = conn.execute("""UPDATE hosted_room_driver_leases SET expires_at=?, updated_at=? + fenced_update(conn, """UPDATE hosted_room_driver_leases SET expires_at=?, updated_at=? WHERE room_id=? AND gateway_id=? AND process_generation=? AND lease_generation=? AND released_at IS NULL AND expires_at > ?""", - (expires_at, now, lease.room_id, lease.gateway_id, lease.process_generation, lease.lease_generation, now)) - if updated.rowcount != 1: - raise StaleLeaseError("driver lease changed during renewal") + (expires_at, now, lease.room_id, *_run_fence(lease), now), + StaleLeaseError("driver lease changed during renewal")) return dataclasses.replace(lease, expires_at=expires_at, reclaimed=False) @@ -584,7 +559,7 @@ def release_lease(db_path: DbPath, lease: DriverLease, *, clock: Clock) -> dict[ """Release the exact active lease generation idempotently.""" now = _timestamp(clock) with _transaction(db_path) as conn: - _require_lease_authority(conn, lease) + _require_room_authority(conn, lease.room_id, lease.gateway_id, lease.authority_epoch) row = conn.execute(_SELECT_LEASE, (lease.room_id,)).fetchone() if not _lease_row_matches(row, lease): raise StaleLeaseError("driver lease is stale") @@ -610,10 +585,8 @@ def admit_task(db_path: DbPath, identity: TaskIdentity, *, payload: Any, clock: now = _timestamp(clock) with _transaction(db_path) as conn: _load_active_room(conn, identity.room_id) - existing = conn.execute(_SELECT_TASK, (identity.room_id, identity.task_id)).fetchone() + existing = _load_task(conn, identity, required=False) if existing is not None: - if _task_identity_from_row(existing) != identity: - raise TaskConflictError("task_id is already bound to a different turn") if existing["payload_digest"] != payload_digest or existing["payload_json"] != payload_json: raise TaskConflictError("task_id is already bound to a different payload") return _task_from_row(existing, idempotent=True) @@ -626,8 +599,8 @@ def admit_task(db_path: DbPath, identity: TaskIdentity, *, payload: Any, clock: status, execution_generation, cancel_generation, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', 0, 0, ?, ?)""", ( - identity.room_id, identity.task_id, identity.thread_id, identity.turn_id, - normalized_payload["source_event_seq"], payload_json, payload_digest, now, now)) + *dataclasses.astuple(identity), normalized_payload["source_event_seq"], payload_json, payload_digest, + now, now)) return _task_from_row(_load_task(conn, identity)) @@ -655,15 +628,13 @@ def start_task( if next_queued is None or next_queued["task_id"] != identity.task_id: raise InvalidTaskTransitionError("task is not next in the hosted room event order") execution_generation = int(row["execution_generation"]) + 1 - updated = conn.execute("""UPDATE hosted_room_driver_tasks + fenced_update(conn, """UPDATE hosted_room_driver_tasks SET status='running', execution_generation=?, run_gateway_id=?, run_process_generation=?, run_lease_generation=?, started_at=?, updated_at=? WHERE room_id=? AND task_id=? AND status='queued' AND cancel_generation=?""", ( - execution_generation, lease.gateway_id, lease.process_generation, lease.lease_generation, now, now, - identity.room_id, identity.task_id, expected_cancel_generation)) - if updated.rowcount != 1: - raise StaleTaskError("task changed during start") + execution_generation, *_run_fence(lease), now, now, identity.room_id, identity.task_id, + expected_cancel_generation), StaleTaskError("task changed during start")) return TaskAttempt( identity=identity, lease=lease, execution_generation=execution_generation, cancel_generation=expected_cancel_generation) @@ -761,11 +732,8 @@ def requeue_not_admitted_task(db_path: DbPath, attempt: TaskAttempt, *, clock: C now = _timestamp(clock) _check_same_room(attempt.lease, attempt.identity) def replay(row: sqlite3.Row) -> dict[str, Any] | None: - requeued = ( - _generations_match(row, "queued", attempt.execution_generation, attempt.cancel_generation) - and row["run_gateway_id"] is None - and row["run_process_generation"] is None - and row["run_lease_generation"] is None) + requeued = _generations_match(row, "queued", attempt.execution_generation, attempt.cancel_generation) and ( + row["run_gateway_id"], row["run_process_generation"], row["run_lease_generation"]) == (None, None, None) return _task_from_row(row, idempotent=True) if requeued else None return _run_fence_transition( db_path, attempt, guard_stale="not-admitted task attempt lost its fence", @@ -829,7 +797,7 @@ def recover_room(db_path: DbPath, lease: DriverLease, *, clock: Clock) -> dict[s """Fence abandoned running attempts without requeueing uncertain work.""" now = _timestamp(clock) foreign_running = f"room_id=? AND status='running' AND NOT ({_RUN_FENCE})" - fence = (lease.room_id, lease.gateway_id, lease.process_generation, lease.lease_generation) + fence = (lease.room_id, *_run_fence(lease)) with _transaction(db_path) as conn: _require_active_lease(conn, lease, now=now) stale_rows = conn.execute( diff --git a/gateway/hosted_room_execution_policy.py b/gateway/hosted_room_execution_policy.py index b12c5bfae4..b425d850ff 100644 --- a/gateway/hosted_room_execution_policy.py +++ b/gateway/hosted_room_execution_policy.py @@ -30,7 +30,6 @@ def _identifier(value: Any, *, field: str) -> str: @dataclass(frozen=True) class RoomExecutionPolicy: """Immutable target policy applied at the agent and approval boundaries.""" - version: int target_profile: str enabled_toolsets: tuple[str, ...] diff --git a/gateway/hosted_room_links.py b/gateway/hosted_room_links.py index ac24473ce0..000986b20c 100644 --- a/gateway/hosted_room_links.py +++ b/gateway/hosted_room_links.py @@ -1,8 +1,7 @@ """Private SQLite storage for negotiated hosted-room links. -Route metadata and its scoped grant share the gateway's private root -``state.db``. SQLite WAL plus ``BEGIN IMMEDIATE`` owns concurrency; grants are -never included in reprs, status payloads, or exception messages. +Route metadata and its scoped grant share the gateway's private root ``state.db``. SQLite WAL plus +``BEGIN IMMEDIATE`` owns concurrency; grants are never included in reprs, status payloads, or exception messages. """ from __future__ import annotations diff --git a/gateway/hosted_room_peer.py b/gateway/hosted_room_peer.py index 21a005d637..4264ea0ec1 100644 --- a/gateway/hosted_room_peer.py +++ b/gateway/hosted_room_peer.py @@ -1,8 +1,7 @@ """Typed contracts for autonomous cross-gateway hosted-room members. -The Desktop may bootstrap an invitation but is never the issuer or runtime -courier: the target gateway verifies a scoped grant and the full task -coordinates before admitting any model or tool work. +The Desktop may bootstrap an invitation but is never the issuer or runtime courier: the target gateway +verifies a scoped grant and the full task coordinates before admitting any model or tool work. """ from __future__ import annotations @@ -53,8 +52,7 @@ def _gateway_room_grant_secret_for_home(home_value: str) -> bytes: (home := Path(home_value)).mkdir(parents=True, exist_ok=True) path = home / _ROOM_GRANT_SECRET_FILE def _read() -> bytes: - data = path.read_bytes() - if len(data) != 32: + if len(data := path.read_bytes()) != 32: raise HostedRoomGrantError("gateway RoomLink secret is invalid") if stat.S_IMODE(path.stat().st_mode) & 0o077: path.chmod(0o600) @@ -195,7 +193,6 @@ _CATALOG_FIELDS = { @dataclass(frozen=True) class GatewayRoomCatalog: """Authenticated gateway capabilities inherited by its Bots.""" - installation_id: str protocol_versions: tuple[int, ...] link_modes: tuple[LinkMode, ...] @@ -220,8 +217,8 @@ class GatewayRoomCatalog: policy = RoomExecutionPolicy.from_mapping(value["execution_policy"]) endpoint = _parse_endpoint(value["endpoint"]) if "endpoint" in value else (None, None, None) catalog = cls( - installation_id, versions, links, value["persistent_process"], value["text"], value["attachments"], - policy, _digest(value["catalog_digest"], field="catalog_digest"), *endpoint) + 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 @@ -252,7 +249,7 @@ def catalog_mapping( """Build a canonical catalog mapping with its digest.""" # A Desktop-managed gateway exits with the app: the caller's flag is only an upper bound. persistent_process = bool(persistent_process and os.getenv("HERMES_DESKTOP") != "1") - profile = (str(target_profile or "").strip() or (os.getenv("HERMES_PROFILE") or "default").strip() or "default") + profile = str(target_profile or "").strip() or (os.getenv("HERMES_PROFILE") or "default").strip() or "default" checked_policy = RoomExecutionPolicy.from_mapping( execution_policy or execution_policy_mapping(target_profile=profile)) # A RoomLink run is initiated by another installation. Process-wide YOLO mode bypasses the scoped @@ -354,7 +351,6 @@ _DISPATCH_FIELDS: dict[str, Callable[..., Any]] = dict( @dataclass(frozen=True) class HostedMemberDispatch: """Recipient-validated identity for one remote room member attempt.""" - protocol_version: int room_id: str home_install_id: str @@ -380,8 +376,7 @@ class HostedMemberDispatch: @classmethod def from_mapping(cls, value: Mapping[str, Any]) -> "HostedMemberDispatch": _exact_fields(value, required=set(_DISPATCH_FIELDS) | {"prompt", "prompt_digest"}, label="dispatch") - prompt = value["prompt"] - if not isinstance(prompt, str) or not prompt.strip(): + if not isinstance(prompt := value["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) prompt_digest = _digest(value["prompt_digest"], field="prompt_digest") @@ -481,8 +476,7 @@ def decode_room_grant(secret: bytes, token: str, *, permission: str, now: float 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: raise HostedRoomGrantError("room grant is expired or not active") - permissions = payload.get("permissions") - if not isinstance(permissions, list) or permission not in permissions: + if not isinstance(permissions := payload.get("permissions"), list) or permission not in permissions: raise HostedRoomGrantError("room grant does not allow this operation") return payload diff --git a/gateway/hosted_room_policy_checkpoint.py b/gateway/hosted_room_policy_checkpoint.py index e1a863cc99..9def25f146 100644 --- a/gateway/hosted_room_policy_checkpoint.py +++ b/gateway/hosted_room_policy_checkpoint.py @@ -1,8 +1,8 @@ """Durable bounded policy projection for hosted Group Chat preparation. -The append-only room log remains the user-visible source of truth. This module -materializes only the state needed to choose and reconstruct the next active -discussion, so a busy room does not replay its complete history every poll. +The append-only room log remains the user-visible source of truth. This module materializes only the state +needed to choose and reconstruct the next active discussion, so a busy room does not replay its complete +history every poll. """ from __future__ import annotations @@ -14,7 +14,7 @@ from pathlib import Path from typing import Any, Callable, Mapping from gateway import hosted_rooms -from gateway.hosted_rooms_common import DbPath, compact_json +from gateway.hosted_rooms_common import DbPath, compact_json, fenced_update MAX_ACTIVE_POLICY_EVENTS = 64 @@ -69,7 +69,6 @@ _TRANSCRIPT_EVENTS_SQL = f"""WITH transcript_events(seq) AS ( @dataclass(frozen=True) class PolicySnapshot: """Bounded active policy input at one durable room-log cursor.""" - through_seq: int stopped_through_seq: int events: tuple[dict[str, Any], ...] @@ -100,7 +99,6 @@ def _settled_message( class HostedRoomPolicyCheckpoint: """Incrementally index room policy without compacting visible history.""" - def __init__(self, db_path: DbPath) -> None: self.db_path = Path(db_path) with self._connect() as conn: @@ -146,17 +144,15 @@ class HostedRoomPolicyCheckpoint: """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""", (room_id, through_seq)): - message_event_id = _text(json.loads(row["payload_json"]), "message_event_id") - if message_event_id: - settled_seq_by_message[message_event_id] = int(row["seq"]) - rows = conn.execute( + settled_seq_by_message = { + message_event_id: int(row["seq"]) + 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""", (room_id, through_seq)) + if (message_event_id := _text(json.loads(row["payload_json"]), "message_event_id"))} + for row in conn.execute( 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: + ORDER BY seq""", (room_id, through_seq)): if row["kind"] == "message.member" and row["event_id"] not in settled_seq_by_message: continue event = _event_from_room_row(row) @@ -178,8 +174,7 @@ class HostedRoomPolicyCheckpoint: rows = conn.execute(_TRANSCRIPT_EVENTS_SQL, (room_id, thread_id, room_id, thread_id, room_id)).fetchall() events_by_seq = { int(event["seq"]): event - for event in ( - *(_event_from_room_row(row) for row in rows), *(json.loads(row["event_json"]) for row in active_rows))} + for event in (*map(_event_from_room_row, rows), *(json.loads(row["event_json"]) for row in active_rows))} return [events_by_seq[seq] for seq in sorted(events_by_seq)] # -- per-kind projection handlers (dispatched by _apply_event) ----------- @@ -295,11 +290,10 @@ class HostedRoomPolicyCheckpoint: _require_room(conn, room_id) 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)) - if updated.rowcount != 1: - raise RuntimeError("room policy cursor disappeared during replay") + fenced_update( + conn, "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), + RuntimeError("room policy cursor disappeared during replay")) cursor = next_cursor return cursor diff --git a/gateway/hosted_room_replicas.py b/gateway/hosted_room_replicas.py index 741fc6de36..3a786b2f31 100644 --- a/gateway/hosted_room_replicas.py +++ b/gateway/hosted_room_replicas.py @@ -11,7 +11,7 @@ from __future__ import annotations import json import sqlite3 -from contextlib import contextmanager +from contextlib import closing, contextmanager from functools import partial from typing import Any, Iterator @@ -53,16 +53,10 @@ def _initialize_replica_schema(conn: sqlite3.Connection) -> None: @contextmanager def _replica_transaction(db_path: DbPath) -> Iterator[sqlite3.Connection]: - """Ensure the replica schema (own autocommit connection), then open an IMMEDIATE transaction. - - The DDL is deliberately re-run inside the transaction: that double init is the established statement order. - """ - conn = _connect(db_path) - try: - with conn: - _initialize_replica_schema(conn) - finally: - conn.close() + """Ensure the replica schema (own autocommit connection), then open an IMMEDIATE transaction. The DDL is + deliberately re-run inside the transaction: that double init is the established statement order.""" + with closing(_connect(db_path)) as conn, conn: + _initialize_replica_schema(conn) with _transaction(db_path, immediate=True) as conn: _initialize_replica_schema(conn) yield conn @@ -206,9 +200,9 @@ def promote_replica( ) -> dict[str, Any]: """Continue a replicated room on THIS gateway at ``epoch + 1``. - Copies the replica log into the authoritative store and appends a lineage-proving - ``authority.claimed`` event, so wherever the claim replicates the old epoch is stale and every - fenced primitive rejects it. The caller decides takeover is safe; this makes it atomic and provable. + Copies the replica log into the authoritative store and appends a lineage-proving ``authority.claimed`` + event, so wherever the claim replicates the old epoch is stale and every fenced primitive rejects it. + The caller decides takeover is safe; this makes it atomic and provable. """ room_id = _room_id(room_id) if not isinstance(reason, str) or not reason or len(reason) > 200: @@ -225,10 +219,8 @@ def promote_replica( raise RoomConflictError("room_id already exists in the local authoritative store") if conn.execute("SELECT 1 FROM hosted_room_retired_ids WHERE room_id=?", (room_id,)).fetchone(): raise RoomConflictError("room_id belongs to a disbanded room") - previous_gateway = str(replica["authority_gateway_id"]) - previous_epoch = int(replica["authority_epoch"]) - target_epoch = previous_epoch + 1 - claim_seq = int(replica["last_seq"]) + 1 + previous_gateway, previous_epoch = str(replica["authority_gateway_id"]), int(replica["authority_epoch"]) + target_epoch, claim_seq = previous_epoch + 1, int(replica["last_seq"]) + 1 claim = _control_event("claimed", target_epoch, { "previous_gateway_id": previous_gateway, "authority_gateway_id": local_gateway, "authority_epoch": target_epoch, "promoted_from_replica": True, "reason": reason}) @@ -257,9 +249,9 @@ def demote_room( ) -> dict[str, Any]: """Fence THIS gateway's stale room authority against a proven newer epoch. - When a returning gateway observes (replicated ``authority.claimed`` or a transport rejection) that - another gateway owns the room at a higher epoch, append ``authority.lost`` and adopt the observed - lineage so no local send can commit at the stale epoch. Idempotent per lineage. + When a returning gateway observes (replicated ``authority.claimed`` or a transport rejection) that another + gateway owns the room at a higher epoch, append ``authority.lost`` and adopt the observed lineage so no + local send can commit at the stale epoch. Idempotent per lineage. """ room_id = _room_id(room_id) observed_gateway_id = _validate_identifier( @@ -272,8 +264,7 @@ def demote_room( FROM hosted_rooms WHERE room_id=? AND disbanded_at IS NULL""", (room_id,)).fetchone() if row is None: raise ReplicaError("room not found in the local authoritative store") - current_gateway = str(row["authority_gateway_id"]) - current_epoch = int(row["authority_epoch"]) + current_gateway, current_epoch = str(row["authority_gateway_id"]), int(row["authority_epoch"]) if current_gateway == observed_gateway_id and current_epoch == observed_epoch: return { "room_id": room_id, "authority_gateway_id": current_gateway, "authority_epoch": current_epoch, diff --git a/gateway/hosted_rooms.py b/gateway/hosted_rooms.py index 9ece0865ae..4a823bf0fd 100644 --- a/gateway/hosted_rooms.py +++ b/gateway/hosted_rooms.py @@ -1,9 +1,8 @@ """Durable state for gateway-hosted Bot Mode rooms. -Owns only hosted-room identity and its append-only event log; delivery, relay leasing -and agent turns belong to the relay and the hosted-room driver, so the log composes with -a durable relay without a second transport queue. Callers supply the database path -(production handlers use the gateway's root ``state.db``). +Owns only hosted-room identity and its append-only event log; delivery, relay leasing and agent turns +belong to the relay and the hosted-room driver, so the log composes with a durable relay without a second +transport queue. Callers supply the database path (production handlers use the gateway's root ``state.db``). """ from __future__ import annotations @@ -18,8 +17,8 @@ from pathlib import Path from typing import Any, Mapping from gateway.hosted_rooms_common import ( - DbPath, bounded_int, canonical_json, clock as _now, compact_json, connect, identifier, open_sqlite, table_columns, - table_exists, transaction) + DbPath, bounded_int, canonical_json, clock as _now, compact_json, connect, fenced_update as _fenced_update, + identifier, open_sqlite, table_columns, table_exists, transaction, utf8_len) PROTOCOL_VERSION = 2 MAX_ROOM_ID_CHARS = 128 @@ -39,9 +38,8 @@ MAX_DISBANDED_ROOM_TOMBSTONES = 512 DISBANDED_ROOM_RETENTION_SECONDS = 90 * 24 * 60 * 60 MAX_EVENTS_PER_ROOM = 50_000 MAX_ROOM_EVENT_BYTES = 256 * 1024 * 1024 -# Leave substantial headroom below the pre-update state.db snapshot ceiling. -# Event accounting does not include SQLite indexes or repeated room ids, so the -# logical budget must stay well below the physical-file limit. +# Leave substantial headroom below the pre-update state.db snapshot ceiling: event accounting excludes +# SQLite indexes and repeated room ids, so the logical budget must stay well below the physical-file limit. MAX_GATEWAY_EVENT_BYTES = 16 * 1024 * 1024 CONTROL_EVENT_COUNT_RESERVE = 64 CONTROL_EVENT_BYTE_RESERVE = 1024 * 1024 @@ -62,7 +60,6 @@ _OPTIONAL_ACTOR_FIELDS = ( _ACTOR_FIELDS = frozenset({"kind", "id", *(field for field, _ in _OPTIONAL_ACTOR_FIELDS)}) # --- schema ----------------------------------------------------------------- - _REMOTE_RUN_IDENTITY_COLUMNS = ( "room_id", "home_install_id", "authority_gateway_id", "authority_epoch", "member_id", "target_install_id", "target_profile", "task_id", "execution_generation") @@ -156,7 +153,6 @@ _REQUIRED_COLUMNS = tuple( _REMOTE_RUN_SCHEMA_COLUMNS = _REQUIRED_COLUMNS[4][1] # --- SQL fragments (statement text must stay byte-stable after whitespace normalisation) --- - _EVENT_COLUMNS = ("room_id, seq, event_id, kind, actor_json, authority_epoch, payload_json, created_at") _SELECT_EVENT = f"SELECT {_EVENT_COLUMNS} FROM hosted_room_events WHERE room_id=? AND event_id=?" _INSERT_EVENT = (f"INSERT INTO hosted_room_events ({_EVENT_COLUMNS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?)") @@ -206,7 +202,6 @@ class AuthoritySupersededError(AuthorityConflictError): # --- validation --------------------------------------------------------------- - _canonical_json = partial(canonical_json, error=HostedRoomError, ensure_ascii=False) _validate_identifier = partial(identifier, error=HostedRoomError) _room_id = partial(_validate_identifier, label="room_id", max_chars=MAX_ROOM_ID_CHARS) @@ -338,9 +333,8 @@ def _migrate_remote_run_schema(conn: sqlite3.Connection) -> None: conn.execute("ALTER TABLE hosted_room_remote_runs_migrating RENAME TO hosted_room_remote_runs") -# Draft builds before the actor contract carried no identity. Preserve their -# inert replay rows explicitly as legacy system events rather than guessing a -# user or Bot author. +# Draft builds before the actor contract carried no identity. Preserve their inert replay rows explicitly +# as legacy system events rather than guessing a user or Bot author. _LEGACY_ACTOR_JSON = _system_actor_json("legacy").replace("'", "''") # (table, column, ddl) applied in this exact order; each table's PRAGMA is read on first use. _LEGACY_COLUMN_DDL = ( @@ -380,10 +374,9 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: for statement in _SCHEMA_DDL: conn.execute(statement) _migrate_legacy_columns(conn) - # Old schemas kept the final identity tombstone in hosted_rooms itself. - # Copy those identities before bounded history pruning can remove their - # heavier room/event payloads. This compact registry is intentionally - # permanent: a stale coordinate must never name a different Group Chat. + # Old schemas kept the final identity tombstone in hosted_rooms itself. Copy those identities before + # bounded history pruning can remove their heavier room/event payloads. This compact registry is + # intentionally permanent: a stale coordinate must never name a different Group Chat. conn.execute(_RETIRE_FROM_ROOMS.format(where="disbanded_at IS NOT NULL")) _migrate_remote_run_schema(conn) conn.execute("CREATE INDEX IF NOT EXISTS idx_hosted_room_events_cursor ON hosted_room_events(room_id, seq)") @@ -449,8 +442,7 @@ def _room_row(conn: sqlite3.Connection, sql: str, params: tuple[Any, ...], room_ row = conn.execute(sql, params).fetchone() if row is not None: return row - # A retained disband tombstone still has replayable history; the caller simply - # did not opt into reading disbanded rooms. + # A retained disband tombstone still has replayable history; the caller did not opt into disbanded rooms. retained = conn.execute("SELECT 1 FROM hosted_rooms WHERE room_id=?", (room_id,)).fetchone() if retained is None and _is_retired(conn, room_id): raise RoomHistoryExpiredError("Group Chat history expired; room_id remains permanently retired") @@ -462,12 +454,6 @@ def _require_authority(room: sqlite3.Row, gateway_id: str, epoch: int, message: raise AuthorityConflictError(message) -def _fenced_update(conn: sqlite3.Connection, sql: str, params: tuple, error: Exception) -> None: - """Run a compare-and-swap UPDATE; anything but exactly one row means the fence was lost.""" - if conn.execute(sql, params).rowcount != 1: - raise error - - def _reload(conn: sqlite3.Connection, sql: str, params: tuple, missing: str) -> sqlite3.Row: """Re-read a row this transaction just wrote; a miss is an invariant violation.""" row = conn.execute(sql, params).fetchone() @@ -522,7 +508,7 @@ def _prepare_event( conn: sqlite3.Connection, room: sqlite3.Row, event_id: str, kind: str, actor_json: str, payload_json: str, *, allow_control: bool = False) -> int: """Size one pending event and enforce per-room and gateway capacity; returns its bytes.""" - additional_bytes = len((event_id + kind + actor_json + payload_json).encode("utf-8")) + additional_bytes = utf8_len(event_id, kind, actor_json, payload_json) count_reserve, byte_reserve = (CONTROL_EVENT_COUNT_RESERVE, CONTROL_EVENT_BYTE_RESERVE) if allow_control else (0, 0) gateway_byte_limit = MAX_GATEWAY_EVENT_BYTES + byte_reserve if int(room["next_seq"]) - 1 >= MAX_EVENTS_PER_ROOM + count_reserve: @@ -540,7 +526,6 @@ def _prepare_event( # --- retention ------------------------------------------------------------------- - # Deleted in this order when a disbanded room's payload is purged. _DEPENDENT_TABLES = ( "hosted_room_policy_transcript_state", "hosted_room_policy_transcript", "hosted_room_policy_publications", @@ -710,7 +695,7 @@ def reserve_peer_room( if expiry <= timestamp: raise HostedRoomError("peer room reservation must expire in the future") values = _reservation_claims(claims) - room_id, _member_id, target_profile, gateway_id, epoch = values + room_id, _, target_profile, gateway_id, epoch = values with _transaction(db_path, immediate=True) as conn: conn.execute("DELETE FROM hosted_room_peer_reservations WHERE expires_at<=?", (timestamp,)) authority_rows = conn.execute( @@ -822,11 +807,10 @@ def _adopt_legacy_room( """Claim a 'legacy'-authority room for a real gateway with a fenced claim event.""" target_epoch = int(existing["authority_epoch"]) + 1 seq = int(existing["next_seq"]) - actor_json = _system_actor_json("authority-control") - payload_json = _claim_payload_json("legacy", authority_gateway_id, target_epoch) claim_bytes = _insert_event( - conn, existing, room_id, seq, "system:authority-adopted", "authority.claimed", actor_json, target_epoch, - payload_json, now, allow_control=True) + conn, existing, room_id, seq, "system:authority-adopted", "authority.claimed", + _system_actor_json("authority-control"), target_epoch, + _claim_payload_json("legacy", authority_gateway_id, target_epoch), now, allow_control=True) _fenced_update(conn, """UPDATE hosted_rooms SET members_json=?, authority_gateway_id=?, authority_epoch=?, next_seq=next_seq+1, revision=revision+1, event_bytes=event_bytes+?, updated_at=? @@ -869,8 +853,8 @@ def create_room( if existing["authority_gateway_id"] != authority_gateway_id: raise RoomConflictError("room_id already belongs to a different authority") return _room_from_row(existing, idempotent=True) - active_rooms = int(conn.execute("SELECT COUNT(*) FROM hosted_rooms WHERE disbanded_at IS NULL").fetchone()[0]) - if active_rooms >= MAX_ACTIVE_ROOMS: + active = conn.execute("SELECT COUNT(*) FROM hosted_rooms WHERE disbanded_at IS NULL").fetchone()[0] + if int(active) >= MAX_ACTIVE_ROOMS: raise HostedRoomError("This host has too many active Group Chats. Delete one and try again.") conn.execute( f"""INSERT INTO hosted_rooms ({_ROOM_COLUMNS_WITH_BYTES}) @@ -923,17 +907,14 @@ def rename_room(db_path: DbPath, *, room_id: Any, event_id: Any, name: Any, now: conn.execute(_INSERT_EVENT, ( room_id, seq, event_id, "room.renamed", actor_json, int(room["authority_epoch"]), payload_json, now)) updated = conn.execute(_SELECT_ROOM, (room_id,)).fetchone() - event = _load_event(conn, room_id, event_id) - return {**_room_from_row(updated), "event": _event_from_row(event)} + return {**_room_from_row(updated), "event": _event_from_row(_load_event(conn, room_id, event_id))} def append_event( db_path: DbPath, *, room_id: Any, event_id: Any, kind: Any, actor: Any, payload: Any, authority_gateway_id: Any = None, authority_epoch: Any = None, now: float | None = None) -> dict[str, Any]: - """Append one immutable event and allocate its per-room sequence atomically. - - Repeating an ``event_id`` with identical content returns the original; different content fails closed. - """ + """Append one immutable event and allocate its per-room sequence atomically; repeating an ``event_id`` + with identical content returns the original, different content fails closed.""" room_id = _room_id(room_id) event_id = _event_id(event_id) kind = _validate_event_kind(kind) @@ -984,11 +965,9 @@ def _probe(path: Path, table: str, query: str, params: tuple[Any, ...], unavaila def probe_hosted_room(db_path: DbPath, *, room_id: Any) -> bool: - """Check room ownership without creating or migrating the shared store. - - Runs on the synchronous prompt-admission path for older Desktop clients, so it fails fast - under contention instead of blocking the WebSocket reader for SQLite's ten-second timeout. - """ + """Check room ownership without creating or migrating the shared store; runs on the synchronous + prompt-admission path for older Desktop clients, so it fails fast under contention instead of blocking + the WebSocket reader for SQLite's ten-second timeout.""" return _probe( Path(db_path), "hosted_rooms", "SELECT 1 FROM hosted_rooms WHERE room_id=? AND disbanded_at IS NULL LIMIT 1", (_room_id(room_id),), "hosted room ownership is temporarily unavailable") @@ -1031,11 +1010,9 @@ def request_room_stop( def claim_authority( db_path: DbPath, *, room_id: Any, expected_gateway_id: Any, expected_epoch: Any, new_gateway_id: Any, event_id: Any, now: float | None = None) -> dict[str, Any]: - """Fence a verified authority transfer with a compare-and-swap epoch. - - Does not decide *when* takeover is safe: a replicated driver calls it only after its - lease/quorum policy established that the previous owner can no longer commit. - """ + """Fence a verified authority transfer with a compare-and-swap epoch; does not decide *when* takeover is + safe (a replicated driver calls it only after its lease/quorum policy established that the previous + owner can no longer commit).""" room_id = _room_id(room_id) expected_gateway_id = _actor_id(expected_gateway_id, "expected_gateway_id") new_gateway_id = _actor_id(new_gateway_id, "new_gateway_id") @@ -1110,8 +1087,7 @@ def disband_room( with _transaction(db_path, immediate=True) as conn: room = conn.execute("""SELECT authority_gateway_id, authority_epoch, next_seq, event_bytes, disbanded_at FROM hosted_rooms WHERE room_id=?""", (room_id,)).fetchone() - replay = _disband_replay(conn, room_id, room) - if replay is not None: + if (replay := _disband_replay(conn, room_id, room)) is not None: return replay _require_authority(room, expected_gateway_id, expected_epoch, "stale hosted room authority") disband_bytes = _insert_event( @@ -1169,19 +1145,16 @@ def read_events( cursor = page_events[-1]["seq"] if page_events else since_seq return {"events": page_events, "cursor": cursor, "latest_seq": latest_seq, "has_more": cursor < latest_seq, "authority": authority} - def page_bytes(page: dict[str, Any]) -> int: - return len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) - page = build_page(events) - if events and page_bytes(page) > MAX_LOG_PAGE_BYTES: + def fits(page_events: list[dict[str, Any]]) -> bool: + page_json = json.dumps(build_page(page_events), ensure_ascii=False, separators=(",", ":")) + return utf8_len(page_json) <= MAX_LOG_PAGE_BYTES + if events and not fits(events): # Binary-search the largest prefix whose serialized page fits the budget. low, high = 1, len(events) while low < high: middle = (low + high + 1) // 2 - if page_bytes(build_page(events[:middle])) <= MAX_LOG_PAGE_BYTES: - low = middle - else: - high = middle - 1 - page = build_page(events[:low]) - if page_bytes(page) > MAX_LOG_PAGE_BYTES: + low, high = (middle, high) if fits(events[:middle]) else (low, middle - 1) + events = events[:low] + if not fits(events): raise HostedRoomError("hosted room event exceeds replay page limit") - return page + return build_page(events) diff --git a/gateway/hosted_rooms_common.py b/gateway/hosted_rooms_common.py index 259190b485..dde9b828aa 100644 --- a/gateway/hosted_rooms_common.py +++ b/gateway/hosted_rooms_common.py @@ -139,6 +139,12 @@ def connect( return conn +def fenced_update(conn: sqlite3.Connection, sql: str, params: tuple, error: Exception) -> None: + """Run a compare-and-swap UPDATE; anything but exactly one row means the fence was lost.""" + if conn.execute(sql, params).rowcount != 1: + raise error + + def table_exists(conn: sqlite3.Connection, table: str) -> bool: row = conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone() return row is not None