"""Passive Mail stream ownership for one AgentsServer runtime. This module is dormant unless its owner explicitly enables the next-beta transport. One Member upstream serves all local subscribers. It owns a bounded dedicated reader thread, never an interactive executor or request lease. A failed upstream closes that cohort; client transport reconnects request a fresh scalar proof. There is no reconnect timer, Inbox read, or background refresh. """ from __future__ import annotations from contextlib import suppress import threading from typing import Any from agentsdock_team_hub.mail_hints import MailArrival, MailHintBroker, MailHintClosed from agentsdock_team_hub.mail_hint_streams import MailHintLease, MailHintScopeChanged, MailHintUnsupported, owned_mail_snapshot from agentsdock_team_hub.notification_hints import NotificationBroker, NotificationCursor, NotificationLease, owned_notification_snapshot class _MemberFeed: def __init__(self, owner: RuntimeMailHints, realm: dict[str, Any]) -> None: self.owner, self.realm = owner, dict(realm) self.broker = MailHintBroker(max_subscriptions=16, max_per_recipient=16) self.notifications = NotificationBroker(max_subscriptions=16, max_per_recipient=16) self.version = 2 if owner._member_negotiated(realm, version=2) else 1 self.ready = threading.Event() self.guard = threading.RLock() self.stream = None self.snapshot = None self.closed = False self.close_code = 1012 self.thread = threading.Thread(target=self._read, name="agentsdock-mail-upstream", daemon=True) def _read(self) -> None: try: opener = (self.owner.runtime.client.open_notification_hint_stream if self.version == 2 else self.owner.runtime.client.open_mail_hint_stream) stream = opener(self.realm["connection_id"]) with self.guard: if self.closed: stream.close() return self.stream = stream first = True previous_mail = previous_bulletin = None while True: frame = stream.read() if frame is None: break cursor = (NotificationCursor.from_dict(frame["cursor"]) if self.version == 2 else MailArrival.from_dict(frame["cursor"])) mail = cursor.mail if self.version == 2 else cursor if frame["hub_id"] != self.realm["hub_id"] or mail.team_id != self.realm["team_id"]: raise MailHintClosed("Member Mail stream changed realm") if first: if frame["type"] != "snapshot": raise MailHintClosed("Member Mail stream omitted snapshot") with self.guard: self.snapshot = dict(frame["cursor"]) first = False elif frame["type"] != "hint" or (self.version == 1 and frame["cursor"].get("reset") is not False) or ( self.version == 2 and any(frame["cursor"][lane].get("reset") is not False for lane in ("mail", "bulletin"))): raise MailHintClosed("Member Mail stream changed generation") if mail != previous_mail: self.broker.publish(mail) if self.version == 2: if mail != previous_mail: self.notifications.publish_mail(mail) if cursor.bulletin != previous_bulletin: self.notifications.publish_bulletin(cursor.bulletin) previous_bulletin = cursor.bulletin previous_mail = mail self.ready.set() except Exception as exc: # The local sockets close without serializing peer errors/tokens. if getattr(exc, "status_code", None) in (401, 403): self.close_code = 4403 elif getattr(exc, "status_code", None) in (404, 501): self.close_code = 4406 elif isinstance(exc, ValueError) or getattr(exc, "code", None) == "remote_invalid": self.close_code = 1008 finally: self.close() def close(self) -> None: with self.guard: self.closed = True stream, self.stream = self.stream, None self.broker.close() self.notifications.close() self.ready.set() if stream is not None: with suppress(Exception): stream.close() class RuntimeMailHints: def __init__(self, runtime: Any, *, enabled: bool = False) -> None: self.runtime = runtime self.enabled = enabled is True self.guard = threading.RLock() self.leases: set[MailHintLease | NotificationLease] = set() self.pending = 0 self.member: _MemberFeed | None = None self._scope: dict[str, Any] | None = None self._generation = 0 def _member_negotiated(self, realm: dict[str, Any], *, version: int = 1) -> bool: # The client getter is a bounded in-memory authenticated health receipt, # not discovery I/O. Missing/expired/other-certificate receipts fail shut. getter = getattr(self.runtime.client, "notification_hint_capability" if version == 2 else "mail_hint_capability", None) if not callable(getter): return False try: return getter(realm["connection_id"], realm["certificate_fingerprint"]) is True except Exception: return False def _capture(self, team_id: str | None = None, *, version: int = 1) -> dict[str, Any]: if not self.enabled or self.runtime._completion_closing: raise MailHintClosed("Mail notifications are unavailable") realms = self.runtime.team_realms() choices = [realm for realm in realms if team_id is None or realm["team_id"] == team_id] if len(choices) != 1: raise MailHintScopeChanged("Mail notification realm is unavailable") with self.guard: generation = self._generation realm = {**choices[0], "epoch": self.runtime._team_authority_epoch, "generation": generation, "admission_epoch": getattr(self.runtime, "_host_admission_epoch", 0)} # Cache only metadata even if capability is not negotiated yet. Later # health renders can inspect the receipt without another realm DB read. with self.guard: if realm["generation"] == self._generation: self._scope = realm if realm["realm"] == "secure_peer" and not self._member_negotiated(realm, version=version): raise MailHintUnsupported("Member Mail notifications are not negotiated") return realm def capability(self, *, version: int = 1) -> dict[str, Any]: result = {"enabled": False, "version": version, "websocket_path": "/api/team-mail-hints/events", "websocket_protocol": f"agentsdock.team-mail-hints.v{version}", "mailbox_coverage": True, "mailbox": None} if version == 2: result["bulletin_coverage"] = True if not self.enabled: return result try: # Metadata discovery is cached, not an arrival query in health. with self.guard: scope = self._scope if scope is None: scope = self._capture(version=version) with self.guard: if scope["generation"] != self._generation: return result self._scope = scope if scope["realm"] == "secure_peer" and not self._member_negotiated(scope, version=version): return result result.update(enabled=True, mailbox={"hub_id": scope["hub_id"], "team_id": scope["team_id"], "recipient_server_id": None}) except Exception: pass return result def _authorize(self, realm: dict[str, Any], store=None, *, version: int = 1) -> None: runtime = self.runtime if (not self.enabled or runtime._completion_closing or runtime._team_authority_epoch != realm["epoch"] or realm["generation"] != self._generation): raise MailHintClosed("Mail notification authority changed") if realm["realm"] == "host": if (runtime._host_admission_closed or runtime._hub_store is not store or store.hub_id != realm["hub_id"] or runtime._host_admission_epoch != realm["admission_epoch"]): raise MailHintClosed("Host Mail admission closed") else: if runtime._host_role_active: raise MailHintClosed("Member Mail authority changed") if not self._member_negotiated(realm, version=version): raise MailHintUnsupported("Member Mail notifications are not negotiated") active = runtime._require_active_proxy_connection(realm["connection_id"]) if (active.get("status") != "connected" or "teamspace.read" not in active.get("scopes", ()) or any(active.get(key) != realm.get(key) for key in ( "hub_id", "team_id", "host_server_identity", "certificate_fingerprint"))): raise MailHintClosed("Member Mail binding changed") def subscribe(self, team_id: str, previous_cursor=None, *, version: int = 1) -> MailHintLease | NotificationLease: if type(version) is not int or version not in (1, 2): raise ValueError("Unsupported notification version") realm = self._capture(team_id, version=version) with self.guard: if self.pending + len(self.leases) >= 16: raise MailHintClosed("Local Mail stream capacity reached") self.pending += 1 subscription = lease = None feed = None try: if realm["realm"] == "host": store = self.runtime._hub_store self._authorize(realm, store, version=version) claims = store.local_agent_mail_claims(team_id) owned = owned_notification_snapshot if version == 2 else owned_mail_snapshot _owned, retained = owned(store, claims, team_id, previous_cursor) subscriber = store.subscribe_team_notifications if version == 2 else store.subscribe_team_mail_arrivals subscription, snapshot = subscriber(claims, team_id, previous_cursor=retained) def authorize(): self._authorize(realm, store, version=version) live = store.team_mail_arrival_snapshot(store.local_agent_mail_claims(team_id), team_id) mail_snapshot = snapshot["mail"] if version == 2 else snapshot if live["recipient_server_id"] != mail_snapshot["recipient_server_id"]: raise MailHintClosed("Host Mail recipient changed") expires_at = None # Local control is role-owned, not a bearer lease. else: self._authorize(realm, version=version) with self.guard: feed = self.member if feed is not None and not feed.closed and feed.realm == realm and version > feed.version: # Retire an older negotiated cohort before opening its # replacement. Existing v1 sockets reconnect onto the # same v2 feed; never keep both upstreams alive. feed.close() if feed is None or feed.closed: feed = _MemberFeed(self, realm) self.member = feed feed.thread.start() elif feed.realm != realm or version > feed.version: raise MailHintClosed("Member Mail generation changed") if not feed.ready.wait(timeout=15) or feed.closed or feed.snapshot is None: if feed.close_code == 4406: raise MailHintUnsupported("Member Mail notifications are unsupported") if feed.close_code == 4403: raise MailHintScopeChanged("Member Mail authority is unavailable") if feed.close_code == 1008: raise ValueError("Member Mail protocol is unavailable") raise MailHintClosed("Member Mail stream unavailable") # Subscribe BEFORE the per-desktop scalar proof. An upstream # cursor alone cannot validate this desktop's retained anchor. feed_mail = feed.snapshot["mail"] if feed.version == 2 else feed.snapshot broker = feed.notifications if version == 2 else feed.broker subscription = broker.subscribe(team_id, feed_mail["recipient_server_id"]) snapshotter = (self.runtime.client.team_notification_hint_snapshot if version == 2 else self.runtime.client.team_mail_hint_snapshot) result = snapshotter(realm["connection_id"], previous_cursor) snapshot = result["cursor"] mail_snapshot = snapshot["mail"] if version == 2 else snapshot if result["hub_id"] != realm["hub_id"] or mail_snapshot["recipient_server_id"] != subscription.mailbox[1]: raise MailHintClosed("Member Mail snapshot changed realm") def authorize(): self._authorize(realm, version=version) if feed.closed or self.member is not feed: raise MailHintClosed("Member Mail upstream closed") active = self.runtime._require_active_proxy_connection(realm["connection_id"]) expires_at = float(active["certificate_expires_at"]) def retired(): with self.guard: self.leases.discard(lease) last = not self.leases and not self.pending old = self.member if last else None if last: self.member = None if old is not None: old.close() lease_type = NotificationLease if version == 2 else MailHintLease lease = lease_type(subscription, snapshot, hub_id=realm["hub_id"], authorize=authorize, expires_at=expires_at, on_close=retired, close_code=(lambda: feed.close_code) if feed is not None else None) with self.guard: self.leases.add(lease) lease.revalidate() return lease except BaseException: if lease is not None: lease.close() elif subscription is not None: subscription.close() raise finally: with self.guard: self.pending -= 1 if not self.pending and not self.leases: old, self.member = self.member, None else: old = None if old is not None: old.close() def invalidate(self) -> None: with self.guard: self._generation += 1 self._scope = None leases = tuple(self.leases) old, self.member = self.member, None if old is not None: old.close() for lease in leases: lease.close()