Coverage for core / peer_link / ui_commands.py: 100.0%
28 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Agentic UI Commands — single source for all HARTOS→phone UI dispatches.
4ANY HARTOS agent (speech therapy, marketing, learning, recipe, fleet, etc.)
5can use these helpers to drive the phone UI. All calls go through the
6existing fleet.command MessageBus topic family — the single dispatch channel
7for every device command (TTS, consent, navigation, overlays, game dispatch).
9Topic selection (automatic based on targeting):
10 - device_id provided → 'fleet.command' (per-device topic)
11 - device_id omitted → 'fleet.command.user' (fan-out to all user's devices)
13Single responsibility:
14 - Formats the payload with the correct cmd_type so the RN handler's
15 registry can route to the matching handler in fleetCommandHandler.js.
16 - Selects the right topic template (per-device vs user-fanout) based on
17 whether the caller specified a device_id.
18 - Does NOT own its own transport or state — delegates to
19 get_message_bus().publish(...).
21No parallel paths:
22 - Uses the same fleet.command topic family already subscribed by
23 AutobahnConnectionManager.
24 - Uses the same FCM fallback already handled by MyFirebaseMessagingService.
25 - Uses the same HTTP polling fallback in pollFleetCommands.
27IMPORTANT — when NOT to use ui_overlay_show:
28 Consent-class interactions (grant microphone access, approve agent action,
29 etc.) MUST use the cmd_type 'agent_consent' instead of 'ui_overlay_show'.
30 The agent_consent path has countdown, auto-deny, ack semantics, TV D-pad
31 support, and accessibility — none of which the generic overlay has. A
32 helper for agent_consent lives in this module's future work; until then,
33 call get_message_bus().publish('fleet.command', {cmd_type: 'agent_consent',
34 ...}) directly.
36Layout theming:
37 Layouts passed to ui_overlay_show are rendered via SocialLiquidUI → which
38 wraps ServerDrivenUI with socialTokens theme. If your agent needs a
39 different theme, inject theme tokens via the layout's style.
41Usage:
42 from core.peer_link.ui_commands import ui_navigate, ui_overlay_show
44 # Targeting a specific device (exact delivery)
45 cid = ui_navigate(user_id='10077', device_id='dev-abc',
46 screen='KidsGame', params={'gameId': 'abc'})
47 if cid is None:
48 log.warning("UI command failed to publish")
50 # Fan-out to all of user's devices (phone + tablet + watch)
51 ui_overlay_show(
52 user_id='10077',
53 layout={'type': 'card', 'children': [...]},
54 data={'gameTitle': 'Balloon Pop'},
55 agent_name='SpeechTherapyAgent',
56 )
57"""
59import logging
60import uuid
61from typing import Any, Dict, Optional
63logger = logging.getLogger('hevolve.ui_commands')
66# ─── cmd_type constants (single source of truth, mirror RN COMMAND_HANDLERS) ──
67# If you add a new cmd_type here, add the matching handler in:
68# services/fleetCommandHandler.js → COMMAND_HANDLERS registry
69CMD_UI_NAVIGATE = 'ui_navigate'
70CMD_UI_OVERLAY_SHOW = 'ui_overlay_show'
71CMD_UI_OVERLAY_DISMISS = 'ui_overlay_dismiss'
74def _publish(
75 cmd_type: str,
76 params: Dict[str, Any],
77 user_id: str,
78 device_id: str = '',
79 command_id: Optional[str] = None,
80) -> Optional[str]:
81 """Internal: format and publish a fleet.command for the UI bridge.
83 Returns:
84 The command_id on success, or None if the publish failed.
85 Callers MUST check the return value — a None means the phone
86 will never receive this command.
87 """
88 if not user_id:
89 raise ValueError("user_id is required for UI commands (per-user scoping)")
91 cid = command_id or f"ui-{uuid.uuid4().hex[:12]}"
92 payload = {
93 'cmd_type': cmd_type,
94 'id': cid,
95 **params,
96 }
98 # Select topic: per-device when device_id specified, else user-fanout
99 topic = 'fleet.command' if device_id else 'fleet.command.user'
101 try:
102 from core.peer_link.message_bus import get_message_bus
103 get_message_bus().publish(
104 topic,
105 payload,
106 user_id=user_id,
107 device_id=device_id,
108 )
109 logger.debug("UI command published: %s topic=%s id=%s user=%s device=%s",
110 cmd_type, topic, cid, user_id, device_id or '*')
111 return cid
112 except Exception as e:
113 logger.warning("UI command publish failed (%s topic=%s): %s",
114 cmd_type, topic, e)
115 return None
118def ui_navigate(
119 user_id: str,
120 screen: str,
121 params: Optional[Dict[str, Any]] = None,
122 device_id: str = '',
123 command_id: Optional[str] = None,
124) -> Optional[str]:
125 """Navigate the phone to a specific screen.
127 The RN handler validates the screen name against an allowlist — screens
128 not in the allowlist will be rejected with an ack failure. The allowlist
129 is maintained in services/fleetCommandHandler.js (NAVIGATION_ALLOWLIST).
131 Args:
132 user_id: User whose device(s) should receive the command.
133 screen: Screen name as registered in home.routes.js (e.g. 'KidsGame',
134 'Encounters', 'CommunityDetail').
135 params: Screen params passed to React Navigation (e.g. {'gameId': 'abc'}).
136 device_id: Target specific device. Empty string = fan out to all
137 user's devices via 'fleet.command.user' topic.
138 command_id: Optional caller-supplied id for ack correlation.
140 Returns:
141 The command_id on successful publish, or None on failure.
142 """
143 if not screen:
144 raise ValueError("screen is required for ui_navigate")
146 return _publish(
147 CMD_UI_NAVIGATE,
148 {'screen': screen, 'params': params or {}},
149 user_id=user_id,
150 device_id=device_id,
151 command_id=command_id,
152 )
155def ui_overlay_show(
156 user_id: str,
157 layout: Dict[str, Any],
158 data: Optional[Dict[str, Any]] = None,
159 agent_name: str = 'Agent',
160 device_id: str = '',
161 command_id: Optional[str] = None,
162) -> Optional[str]:
163 """Float an overlay on the phone with a ServerDrivenUI layout.
165 The layout JSON is rendered by the RN app's LiquidOverlay →
166 SocialLiquidUI → ServerDrivenUI chain. SocialLiquidUI injects
167 socialTokens theme, so layouts pick up the social color palette by
168 default. The raw JSON fallback in LiquidOverlay handles layouts that
169 fail to render through SocialLiquidUI.
171 NOT FOR CONSENT: see module docstring. Use 'agent_consent' cmd_type
172 instead for permission / approval flows.
174 NOT SHOWN ON TV: the RN handler rejects this on TV devices (the
175 LiquidOverlay component is not mounted on TV per CommunityView.js).
176 Use ui_navigate to a dedicated screen for TV agents.
178 Args:
179 user_id: User whose device(s) should receive the command.
180 layout: ServerDrivenUI JSON descriptor (type, props, style, children,
181 bind, action, visible). Max depth 20. Must be a dict with a
182 'type' field.
183 data: Context object for data bindings (e.g. {'gameTitle': 'Pong'}).
184 agent_name: Display name shown in the overlay header.
185 device_id: Target specific device. Empty string = fan out to all
186 user's devices via 'fleet.command.user' topic.
187 command_id: Optional caller-supplied id for ack correlation.
189 Returns:
190 The command_id on successful publish, or None on failure.
191 """
192 if not isinstance(layout, dict) or 'type' not in layout:
193 raise ValueError("layout must be a dict with a 'type' field (ServerDrivenUI schema)")
195 return _publish(
196 CMD_UI_OVERLAY_SHOW,
197 {
198 'layout': layout,
199 'data': data or {},
200 'agent_name': agent_name,
201 },
202 user_id=user_id,
203 device_id=device_id,
204 command_id=command_id,
205 )
208def ui_overlay_dismiss(
209 user_id: str,
210 device_id: str = '',
211 command_id: Optional[str] = None,
212) -> Optional[str]:
213 """Dismiss the active overlay on the phone.
215 Args:
216 user_id: User whose device(s) should receive the command.
217 device_id: Target specific device. Empty string = fan out to all
218 user's devices via 'fleet.command.user' topic.
219 command_id: Optional caller-supplied id for ack correlation.
221 Returns:
222 The command_id on successful publish, or None on failure.
223 """
224 return _publish(
225 CMD_UI_OVERLAY_DISMISS,
226 {},
227 user_id=user_id,
228 device_id=device_id,
229 command_id=command_id,
230 )