ed
1c565da7a0
PR2 of the test_full_live_workflow_imgui_assert fix sequence.
When an ImGui scope mismatch (IM_ASSERT(Missing End())) fires in
immapp.run (e.g. after cumulative state corruption from prior sims'
panel renders), the RuntimeError propagates out of app.run(). The
controller's _io_pool gets shut down via __del__/finalization. The
hook server (separate ThreadingHTTPServer) survives. Subsequent test
clicks fail with 'cannot schedule new futures after shutdown' and
the test times out after 120s with no clear signal of what went
wrong.
This commit:
1. Wraps immapp.run in try/except RuntimeError in gui_2.py:618.
On assertion: logs the error to stderr (NOT silent), records
it on controller._gui_degraded_reason and _last_imgui_assert,
and returns from run() so the hook server keeps serving.
2. Adds _gui_degraded_reason and _last_imgui_assert to
AppController.__init__ (initialized to None).
3. Adds /api/gui_health endpoint in api_hooks.py:148. Returns
{healthy, degraded_reason, last_assert, io_pool_alive}.
4. Adds ApiHookClient.get_gui_health() with the matching unit
tests (3 mocked tests + 1 live test).
Per user feedback 2026-06-08:
- The wrap does NOT silently swallow the error. It logs at ERROR
level and surfaces it via the health endpoint.
- Tests can call client.get_gui_health() to detect a degraded GUI
and fail fast with a clear message.
TDD: tests written first, confirmed to fail, then fix applied.
34/34 unit tests pass. 1/1 live test passes (live_gui health
endpoint reports healthy=True on fresh subprocess).
578 lines
37 KiB
Python
578 lines
37 KiB
Python
"""
|
|
API Hook Client - Python client for the Hook API.
|
|
|
|
This module provides a Python client for interacting with the Hook API exposed by the application on port 8999.
|
|
It is used for:
|
|
- Automated GUI testing via the `live_gui` pytest fixture
|
|
- External tool integration
|
|
- Remote control of the application
|
|
|
|
Architecture:
|
|
- Uses requests library for HTTP communication
|
|
- All methods return dict[str, Any] or None
|
|
- Handles connection errors gracefully (returns None on failure)
|
|
|
|
Key Method Categories:
|
|
1. Connection: wait_for_server, get_status
|
|
2. State Query: get_project, get_session. get_performance, get_mma_status
|
|
3. GUI Manipulation: click, set_value, select_tab, select_list_item
|
|
4. Polling: wait_for_event
|
|
5. HITL: request_confirmation
|
|
|
|
Timeout Handling:
|
|
- Standard operations: 5s timeout
|
|
- HITL dialogs: 60s timeout (waits for human input)
|
|
|
|
Integration:
|
|
- Used by simulation tests (tests/visual_sim_mma_v2.py)
|
|
- Used by external tools for automation
|
|
|
|
See Also:
|
|
- src/api_hooks.py for the server implementation
|
|
- docs/guide_tools.md for Hook API documentation
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import requests # type: ignore[import-untyped]
|
|
import sys
|
|
import time
|
|
|
|
from typing import Any
|
|
|
|
|
|
class ApiHookClient:
|
|
|
|
def __init__(self, base_url: str = "http://127.0.0.1:8999", api_key: str | None = None):
|
|
"""
|
|
[C: src/mcp_client.py:_DDGParser.__init__, src/mcp_client.py:_TextExtractor.__init__]
|
|
"""
|
|
self.base_url = base_url.rstrip('/')
|
|
self.api_key = api_key
|
|
|
|
def _make_request(self, method: str, path: str, data: dict | None = None, timeout: float = 5.0) -> dict[str, Any] | None:
|
|
"""
|
|
Helper to make HTTP requests to the hook server.
|
|
[C: tests/test_api_hook_client.py:test_unsupported_method_error]
|
|
"""
|
|
url = f"{self.base_url}{path}"
|
|
headers = {}
|
|
if self.api_key:
|
|
headers["X-API-KEY"] = self.api_key
|
|
if method not in ('GET', 'POST', 'DELETE'):
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
try:
|
|
if method == 'GET': response = requests.get(url, headers=headers, timeout=timeout)
|
|
elif method == 'POST': response = requests.post(url, json=data, headers=headers, timeout=timeout)
|
|
elif method == 'DELETE': response = requests.delete(url, headers=headers, timeout=timeout)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
sys.stderr.write(f"[DEBUG Client] Request failed: {method} {path} - status={response.status_code}\n")
|
|
sys.stderr.flush()
|
|
return None
|
|
except Exception as e:
|
|
sys.stderr.write(f"[DEBUG Client] Request error: {method} {path} - {e}\n")
|
|
sys.stderr.flush()
|
|
return None
|
|
|
|
def wait_for_server(self, timeout: int = 15) -> bool:
|
|
"""
|
|
Polls the health endpoint until the server responds or timeout occurs.
|
|
[C: simulation/live_walkthrough.py:main, simulation/ping_pong.py:main, simulation/sim_base.py:BaseSimulation.setup, tests/smoke_status_hook.py:test_status_hook, tests/test_ai_settings_layout.py:test_change_provider_via_hook, tests/test_ai_settings_layout.py:test_set_params_via_custom_callback, tests/test_auto_switch_sim.py:test_auto_switch_sim, tests/test_conductor_api_hook_integration.py:test_conductor_integrates_api_hook_client_for_verification, tests/test_deepseek_infra.py:test_gui_provider_list_via_hooks, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_extended_sims.py:test_context_sim_live, tests/test_extended_sims.py:test_execution_sim_live, tests/test_extended_sims.py:test_tools_sim_live, tests/test_external_editor_gui.py:test_button_click_is_received, tests/test_external_editor_gui.py:test_patch_modal_shows_with_configured_editor, tests/test_external_editor_gui.py:test_vscode_launches_with_diff_view, tests/test_gui2_parity.py:test_gui2_click_hook_works, tests/test_gui2_parity.py:test_gui2_custom_callback_hook_works, tests/test_gui2_parity.py:test_gui2_set_value_hook_works, tests/test_gui_context_presets.py:test_gui_context_preset_save_load, tests/test_hooks.py:test_live_hook_server_responses, tests/test_live_workflow.py:test_full_live_workflow, tests/test_mma_concurrent_tracks_sim.py:test_mma_concurrent_tracks_execution, tests/test_mma_concurrent_tracks_stress_sim.py:test_mma_concurrent_tracks_stress, tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow, tests/test_patch_modal_gui.py:test_patch_apply_modal_workflow, tests/test_patch_modal_gui.py:test_patch_modal_appears_on_trigger, tests/test_phase6_simulation.py:test_ast_inspector_modal_opens, tests/test_phase6_simulation.py:test_batch_operations_shift_click, tests/test_phase6_simulation.py:test_slice_editor_add_remove, tests/test_preset_windows_layout.py:test_api_hook_under_load, tests/test_preset_windows_layout.py:test_preset_windows_opening, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_rag_visual_sim.py:test_rag_full_lifecycle_sim, tests/test_rag_visual_sim.py:test_rag_settings_persistence_sim, tests/test_selectable_ui.py:test_selectable_label_stability, tests/test_system_prompt_sim.py:test_system_prompt_sim, tests/test_tool_management_layout.py:test_tool_management_gettable_fields, tests/test_tool_management_layout.py:test_tool_management_state_updates, tests/test_ui_cache_controls_sim.py:test_ui_cache_controls, tests/test_undo_redo_sim.py:test_undo_redo_context_mutation, tests/test_undo_redo_sim.py:test_undo_redo_discussion_mutation, tests/test_undo_redo_sim.py:test_undo_redo_lifecycle, tests/test_visual_mma.py:test_visual_mma_components, tests/test_visual_orchestration.py:test_mma_epic_lifecycle, tests/test_visual_sim_gui_ux.py:test_gui_track_creation, tests/test_visual_sim_gui_ux.py:test_gui_ux_event_routing, tests/test_visual_sim_mma_v2.py:test_mma_complete_lifecycle, tests/test_workspace_profiles_sim.py:test_workspace_profiles_restoration, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout]
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
status = self.get_status()
|
|
if status and (status.get("status") == "ok" or "status" in status):
|
|
return True
|
|
time.sleep(0.5)
|
|
return False
|
|
|
|
def get_status(self) -> dict[str, Any]:
|
|
"""
|
|
Checks the health of the hook server.
|
|
[C: tests/test_api_hook_client.py:test_get_status_success, tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_hooks.py:test_live_hook_server_responses, tests/test_mma_concurrent_tracks_stress_sim.py:test_mma_concurrent_tracks_stress, tests/test_phase6_simulation.py:test_ast_inspector_modal_opens, tests/test_phase6_simulation.py:test_batch_operations_shift_click, tests/test_phase6_simulation.py:test_slice_editor_add_remove, tests/test_preset_windows_layout.py:make_request, tests/test_preset_windows_layout.py:test_preset_windows_opening, tests/test_ui_cache_controls_sim.py:test_ui_cache_controls]
|
|
"""
|
|
res = self._make_request('GET', '/status')
|
|
if res is None:
|
|
# For backward compatibility with tests expecting ConnectionError
|
|
# But our _make_request handles it. Let's return empty if failed.
|
|
return {}
|
|
return res
|
|
|
|
def post_session(self, session_entries: list[dict]) -> dict[str, Any]:
|
|
"""
|
|
Updates the session history.
|
|
[C: tests/test_gui_stress_performance.py:test_comms_volume_stress_performance, tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
return self._make_request('POST', '/api/session', data={"session": {"entries": session_entries}}) or {}
|
|
|
|
def get_events(self) -> list[dict[str, Any]]:
|
|
"""Retrieves any pending events from the API event queue."""
|
|
res = self._make_request('GET', '/api/events')
|
|
return res.get("events", []) if res else []
|
|
|
|
def clear_events(self) -> list[dict[str, Any]]:
|
|
"""
|
|
Retrieves and clears the event queue.
|
|
[C: simulation/sim_base.py:BaseSimulation.setup]
|
|
"""
|
|
return self.get_events()
|
|
|
|
def wait_for_event(self, event_type: str, timeout: int = 5) -> dict[str, Any] | None:
|
|
"""
|
|
[C: simulation/sim_base.py:BaseSimulation.wait_for_event, simulation/sim_execution.py:ExecutionSimulation.run, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout]
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
events = self.get_events()
|
|
for ev in events:
|
|
if ev.get("type") == event_type or ev.get("event_type") == event_type:
|
|
return ev
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
def post_gui(self, payload: dict) -> dict[str, Any]:
|
|
"""
|
|
Pushes an event to the GUI's AsyncEventQueue via the /api/gui endpoint.
|
|
[C: tests/test_ai_settings_layout.py:test_set_params_via_custom_callback, tests/test_api_hook_client.py:test_post_gui_success, tests/test_gui2_parity.py:test_gui2_custom_callback_hook_works, tests/test_gui2_parity.py:test_gui2_set_value_hook_works]
|
|
"""
|
|
return self._make_request('POST', '/api/gui', data=payload) or {}
|
|
|
|
def push_event(self, action: str, payload: dict) -> dict[str, Any]:
|
|
"""
|
|
Convenience to push a GUI task.
|
|
[C: tests/test_auto_switch_sim.py:test_auto_switch_sim, tests/test_auto_switch_sim.py:trigger_tier, tests/test_external_editor_gui.py:test_button_click_is_received, tests/test_external_editor_gui.py:test_patch_modal_shows_with_configured_editor, tests/test_external_editor_gui.py:test_vscode_launches_with_diff_view, tests/test_gui_context_presets.py:test_gui_context_preset_save_load, tests/test_gui_text_viewer.py:test_text_viewer_state_update, tests/test_patch_modal_gui.py:test_patch_apply_modal_workflow, tests/test_patch_modal_gui.py:test_patch_modal_appears_on_trigger, tests/test_preset_windows_layout.py:test_preset_windows_opening, tests/test_saved_presets_sim.py:test_preset_manager_modal, tests/test_saved_presets_sim.py:test_preset_switching, tests/test_tool_management_layout.py:test_tool_management_state_updates, tests/test_tool_presets_sim.py:test_tool_preset_switching, tests/test_visual_mma.py:test_visual_mma_components, tests/test_visual_sim_gui_ux.py:test_gui_track_creation, tests/test_visual_sim_gui_ux.py:test_gui_ux_event_routing, tests/test_workspace_profiles_sim.py:test_workspace_profiles_restoration, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout]
|
|
"""
|
|
return self.post_gui({"action": action, **payload})
|
|
|
|
#region: Data
|
|
|
|
def get_gui_state(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the full GUI state available via the hook API.
|
|
[C: tests/test_ai_settings_layout.py:test_change_provider_via_hook, tests/test_ai_settings_layout.py:test_set_params_via_custom_callback, tests/test_conductor_api_hook_integration.py:simulate_conductor_phase_completion, tests/test_external_editor_gui.py:test_button_click_is_received, tests/test_external_editor_gui.py:test_patch_modal_shows_with_configured_editor, tests/test_external_editor_gui.py:test_vscode_launches_with_diff_view, tests/test_gui_text_viewer.py:test_text_viewer_state_update, tests/test_hooks.py:test_live_hook_server_responses, tests/test_live_gui_integration_v2.py:test_api_gui_state_live, tests/test_live_workflow.py:test_full_live_workflow, tests/test_live_workflow.py:wait_for_value, tests/test_patch_modal_gui.py:test_patch_apply_modal_workflow, tests/test_patch_modal_gui.py:test_patch_modal_appears_on_trigger, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_saved_presets_sim.py:test_preset_manager_modal, tests/test_saved_presets_sim.py:test_preset_switching, tests/test_task_dag_popout_sim.py:test_task_dag_popout, tests/test_tool_management_layout.py:test_tool_management_gettable_fields, tests/test_tool_management_layout.py:test_tool_management_state_updates, tests/test_tool_presets_sim.py:test_tool_preset_switching, tests/test_usage_analytics_popout_sim.py:test_usage_analytics_popout, tests/test_visual_mma.py:test_visual_mma_components]
|
|
"""
|
|
return self._make_request('GET', '/api/gui/state') or {}
|
|
|
|
def get_value(self, item: str) -> Any:
|
|
"""
|
|
Gets the value of a GUI item via its mapped field.
|
|
[C: simulation/sim_ai_settings.py:AISettingsSimulation.run, simulation/sim_base.py:BaseSimulation.get_value, simulation/sim_base.py:BaseSimulation.setup, simulation/sim_base.py:BaseSimulation.wait_for_element, simulation/sim_context.py:ContextSimulation.run, simulation/sim_execution.py:ExecutionSimulation.run, simulation/workflow_sim.py:WorkflowSimulator.run_discussion_turn_async, simulation/workflow_sim.py:WorkflowSimulator.wait_for_ai_response, tests/smoke_status_hook.py:test_status_hook, tests/smoke_status_hook.py:wait_for_value, tests/test_auto_switch_sim.py:test_auto_switch_sim, tests/test_deepseek_infra.py:test_gui_provider_list_via_hooks, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_gui2_parity.py:test_gui2_click_hook_works, tests/test_gui2_parity.py:test_gui2_set_value_hook_works, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_rag_visual_sim.py:test_rag_full_lifecycle_sim, tests/test_rag_visual_sim.py:test_rag_settings_persistence_sim, tests/test_selectable_ui.py:test_selectable_label_stability, tests/test_system_prompt_sim.py:test_system_prompt_sim, tests/test_undo_redo_sim.py:test_undo_redo_context_mutation, tests/test_undo_redo_sim.py:test_undo_redo_discussion_mutation, tests/test_undo_redo_sim.py:test_undo_redo_lifecycle, tests/test_workspace_profiles_sim.py:test_workspace_profiles_restoration]
|
|
"""
|
|
# Dict-key bracket notation: e.g. 'show_windows["Project Settings"]'
|
|
if "[" in item and item.endswith("]"):
|
|
dict_name, _, key_part = item.partition("[")
|
|
key = key_part[:-1].strip().strip("'\"")
|
|
state = self.get_gui_state()
|
|
if isinstance(state.get(dict_name), dict) and key in state[dict_name]:
|
|
return state[dict_name][key]
|
|
return None
|
|
# Try state endpoint first (new preferred way)
|
|
state = self.get_gui_state()
|
|
if item in state:
|
|
return state[item]
|
|
# Fallback for thinking/live/prior which are in diagnostics
|
|
diag = self.get_gui_diagnostics()
|
|
if diag and item in diag:
|
|
return diag[item]
|
|
# Map common indicator tags to diagnostics keys
|
|
mapping = {
|
|
"thinking_indicator": "thinking",
|
|
"operations_live_indicator": "live",
|
|
"prior_session_indicator": "prior"
|
|
}
|
|
key = mapping.get(item)
|
|
if diag and key and key in diag:
|
|
return diag[key]
|
|
return None
|
|
|
|
def get_text_value(self, item_tag: str) -> str | None:
|
|
"""
|
|
Wraps get_value and returns its string representation, or None.
|
|
[C: tests/test_api_hook_client.py:test_get_text_value]
|
|
"""
|
|
val = self.get_value(item_tag)
|
|
return str(val) if val is not None else None
|
|
|
|
def set_value(self, item: str, value: Any) -> dict[str, Any]:
|
|
"""
|
|
Sets the value of a GUI widget.
|
|
[C: simulation/live_walkthrough.py:main, simulation/ping_pong.py:main, simulation/sim_ai_settings.py:AISettingsSimulation.run, simulation/sim_base.py:BaseSimulation.setup, simulation/workflow_sim.py:WorkflowSimulator.create_discussion, simulation/workflow_sim.py:WorkflowSimulator.run_discussion_turn_async, simulation/workflow_sim.py:WorkflowSimulator.setup_new_project, simulation/workflow_sim.py:WorkflowSimulator.truncate_history, tests/smoke_status_hook.py:test_status_hook, tests/test_ai_settings_layout.py:test_change_provider_via_hook, tests/test_auto_switch_sim.py:test_auto_switch_sim, tests/test_deepseek_infra.py:test_gui_provider_list_via_hooks, tests/test_extended_sims.py:test_ai_settings_sim_live, tests/test_extended_sims.py:test_context_sim_live, tests/test_extended_sims.py:test_execution_sim_live, tests/test_extended_sims.py:test_tools_sim_live, tests/test_gui2_parity.py:test_gui2_click_hook_works, tests/test_gui2_performance.py:test_performance_benchmarking, tests/test_live_gui_integration_v2.py:test_api_gui_state_live, tests/test_live_workflow.py:test_full_live_workflow, tests/test_mma_concurrent_tracks_sim.py:test_mma_concurrent_tracks_execution, tests/test_mma_concurrent_tracks_stress_sim.py:test_mma_concurrent_tracks_stress, tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_rag_visual_sim.py:test_rag_full_lifecycle_sim, tests/test_rag_visual_sim.py:test_rag_settings_persistence_sim, tests/test_saved_presets_sim.py:test_preset_manager_modal, tests/test_selectable_ui.py:test_selectable_label_stability, tests/test_system_prompt_sim.py:test_system_prompt_sim, tests/test_task_dag_popout_sim.py:test_task_dag_popout, tests/test_tool_presets_sim.py:test_tool_preset_switching, tests/test_undo_redo_sim.py:test_undo_redo_context_mutation, tests/test_undo_redo_sim.py:test_undo_redo_discussion_mutation, tests/test_undo_redo_sim.py:test_undo_redo_lifecycle, tests/test_usage_analytics_popout_sim.py:test_usage_analytics_popout, tests/test_visual_mma.py:test_visual_mma_components, tests/test_visual_orchestration.py:test_mma_epic_lifecycle, tests/test_visual_sim_mma_v2.py:test_mma_complete_lifecycle, tests/test_workspace_profiles_sim.py:test_workspace_profiles_restoration, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout]
|
|
"""
|
|
return self.post_gui({"action": "set_value", "item": item, "value": value})
|
|
|
|
#endregion: Data
|
|
|
|
#region: Input
|
|
|
|
def click(self, item: str, user_data: Any = None) -> dict[str, Any]:
|
|
"""
|
|
Simulates a button click.
|
|
[C: simulation/live_walkthrough.py:main, simulation/ping_pong.py:main, simulation/sim_base.py:BaseSimulation.setup, simulation/sim_context.py:ContextSimulation.run, simulation/sim_execution.py:ExecutionSimulation.run, simulation/workflow_sim.py:WorkflowSimulator.create_discussion, simulation/workflow_sim.py:WorkflowSimulator.load_prior_log, simulation/workflow_sim.py:WorkflowSimulator.run_discussion_turn_async, simulation/workflow_sim.py:WorkflowSimulator.setup_new_project, simulation/workflow_sim.py:WorkflowSimulator.truncate_history, simulation/workflow_sim.py:WorkflowSimulator.wait_for_ai_response, tests/test_external_editor_gui.py:test_button_click_is_received, tests/test_external_editor_gui.py:test_vscode_launches_with_diff_view, tests/test_gui2_parity.py:test_gui2_click_hook_works, tests/test_gui_text_viewer.py:test_text_viewer_state_update, tests/test_live_workflow.py:test_full_live_workflow, tests/test_mma_concurrent_tracks_sim.py:test_mma_concurrent_tracks_execution, tests/test_mma_concurrent_tracks_stress_sim.py:test_mma_concurrent_tracks_stress, tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim, tests/test_rag_visual_sim.py:test_rag_full_lifecycle_sim, tests/test_saved_presets_sim.py:test_preset_manager_modal, tests/test_saved_presets_sim.py:test_preset_switching, tests/test_system_prompt_sim.py:test_system_prompt_sim, tests/test_ui_cache_controls_sim.py:test_ui_cache_controls, tests/test_undo_redo_sim.py:test_undo_redo_context_mutation, tests/test_undo_redo_sim.py:test_undo_redo_discussion_mutation, tests/test_undo_redo_sim.py:test_undo_redo_lifecycle, tests/test_visual_mma.py:test_visual_mma_components, tests/test_visual_orchestration.py:test_mma_epic_lifecycle, tests/test_visual_sim_gui_ux.py:test_gui_track_creation, tests/test_visual_sim_gui_ux.py:test_gui_ux_event_routing, tests/test_visual_sim_mma_v2.py:_drain_approvals, tests/test_visual_sim_mma_v2.py:test_mma_complete_lifecycle, tests/test_z_negative_flows.py:test_mock_error_result, tests/test_z_negative_flows.py:test_mock_malformed_json, tests/test_z_negative_flows.py:test_mock_timeout]
|
|
"""
|
|
return self.post_gui({"action": "click", "item": item, "user_data": user_data})
|
|
|
|
def drag(self, src_item: str, dst_item: str) -> dict[str, Any]:
|
|
"""
|
|
Simulates a drag and drop operation.
|
|
[C: tests/test_api_hook_client.py:test_drag_success]
|
|
"""
|
|
return self.push_event("drag", {"src_item": src_item, "dst_item": dst_item})
|
|
|
|
def right_click(self, item: str) -> dict[str, Any]:
|
|
"""
|
|
Simulates a right-click on an item.
|
|
[C: tests/test_api_hook_client.py:test_right_click_success]
|
|
"""
|
|
return self.push_event("right_click", {"item": item})
|
|
|
|
def request_confirmation(self, tool_name: str, args: dict) -> bool | None:
|
|
"""
|
|
Pushes a manual confirmation request and waits for response.
|
|
Blocks for up to 60 seconds.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_sync_hooks.py:test_api_ask_client_error, tests/test_sync_hooks.py:test_api_ask_client_method, tests/test_sync_hooks.py:test_api_ask_client_rejection]
|
|
"""
|
|
# Long timeout as this waits for human input (60 seconds)
|
|
res = self._make_request('POST', '/api/ask',
|
|
data={'type': 'tool_approval', 'tool': tool_name, 'args': args},
|
|
timeout=60.0)
|
|
return res.get('response') if res else None
|
|
|
|
def select_list_item(self, item: str, value: str) -> dict[str, Any]:
|
|
"""
|
|
Selects an item in a listbox or combo.
|
|
[C: simulation/workflow_sim.py:WorkflowSimulator.create_discussion, simulation/workflow_sim.py:WorkflowSimulator.switch_discussion, tests/test_api_hook_extensions.py:test_select_list_item_integration, tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
return self.set_value(item, value)
|
|
|
|
def select_tab(self, item: str, value: str) -> dict[str, Any]:
|
|
"""
|
|
Selects a specific tab in a tab bar.
|
|
[C: simulation/live_walkthrough.py:main, tests/test_api_hook_extensions.py:test_select_tab_integration]
|
|
"""
|
|
return self.set_value(item, value)
|
|
|
|
#endregion: Input
|
|
|
|
#region: Patching
|
|
|
|
def trigger_patch(self, patch_text: str, file_paths: list[str]) -> dict[str, Any]:
|
|
"""Triggers the patch modal to show in the GUI."""
|
|
return self._make_request('POST', '/api/patch/trigger', data={
|
|
"patch_text": patch_text,
|
|
"file_paths": file_paths
|
|
}) or {}
|
|
|
|
def apply_patch(self) -> dict[str, Any]:
|
|
"""
|
|
Applies the pending patch.
|
|
[C: tests/test_patch_modal.py:test_apply_callback]
|
|
"""
|
|
return self._make_request('POST', '/api/patch/apply') or {}
|
|
|
|
def reject_patch(self) -> dict[str, Any]:
|
|
"""
|
|
Rejects the pending patch.
|
|
[C: tests/test_patch_modal.py:test_reject_callback, tests/test_patch_modal.py:test_reject_patch]
|
|
"""
|
|
return self._make_request('POST', '/api/patch/reject') or {}
|
|
|
|
def get_patch_status(self) -> dict[str, Any]:
|
|
"""Gets the current patch modal status."""
|
|
return self._make_request('GET', '/api/patch/status') or {}
|
|
|
|
#endregion: Patching
|
|
|
|
#region: Diagnostics
|
|
|
|
def get_indicator_state(self, item_tag: str) -> dict[str, bool]:
|
|
"""
|
|
Returns the visibility/active state of a status indicator.
|
|
[C: simulation/live_walkthrough.py:main, tests/test_api_hook_extensions.py:test_get_indicator_state_integration, tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
val = self.get_value(item_tag)
|
|
return {"shown": bool(val)}
|
|
|
|
def get_gui_diagnostics(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves performance and diagnostic metrics.
|
|
[C: tests/test_api_hook_client.py:test_get_performance_success, tests/test_hooks.py:test_live_hook_server_responses, tests/test_selectable_ui.py:test_selectable_label_stability, tests/test_visual_sim_gui_ux.py:test_gui_ux_event_routing]
|
|
"""
|
|
return self._make_request('GET', '/api/gui/diagnostics') or {}
|
|
|
|
def get_performance(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves performance metrics from the dedicated endpoint.
|
|
[C: tests/test_gui2_performance.py:test_performance_benchmarking, tests/test_gui_performance_requirements.py:test_idle_performance_requirements, tests/test_gui_stress_performance.py:test_comms_volume_stress_performance, tests/test_selectable_ui.py:test_selectable_label_stability]
|
|
"""
|
|
return self._make_request('GET', '/api/performance') or {}
|
|
|
|
def get_warmup_status(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the current warmup status: {pending, completed, failed}.
|
|
[C: tests/test_api_hooks_warmup.py:test_get_warmup_status_calls_correct_endpoint, tests/test_api_hooks_warmup.py:test_get_warmup_status_handles_empty_response, tests/test_api_hooks_warmup.py:test_live_warmup_status_endpoint]
|
|
"""
|
|
return self._make_request('GET', '/api/warmup_status') or {}
|
|
|
|
def get_warmup_wait(self, timeout: float = 30.0) -> dict[str, Any]:
|
|
"""
|
|
Blocks server-side up to `timeout` seconds waiting for the warmup to
|
|
complete, then returns the final status. Useful for external clients
|
|
that need to wait until the system is fully ready before issuing AI
|
|
requests.
|
|
[C: tests/test_api_hooks_warmup.py:test_get_warmup_wait_passes_timeout_as_query_string, tests/test_api_hooks_warmup.py:test_get_warmup_wait_uses_default_timeout_when_unspecified, tests/test_api_hooks_warmup.py:test_get_warmup_wait_handles_empty_response, tests/test_api_hooks_warmup.py:test_live_warmup_wait_endpoint_completes]
|
|
"""
|
|
return self._make_request('GET', f'/api/warmup_wait?timeout={timeout}') or {}
|
|
|
|
def get_warmup_canaries(self) -> list[dict[str, Any]]:
|
|
"""
|
|
Returns per-module import canary records: list of dicts with
|
|
canary_id, module, thread_name, thread_id, submit_ts, start_ts,
|
|
end_ts, elapsed_ms, status, error. Used for debugging which
|
|
worker thread loaded which module and how long it took.
|
|
[C: tests/test_api_hooks_warmup.py:test_get_warmup_canaries_in_live_gui]
|
|
"""
|
|
result = self._make_request('GET', '/api/warmup_canaries') or {}
|
|
return result.get("canaries", []) if isinstance(result, dict) else []
|
|
|
|
def get_startup_timeline(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the startup timeline: dict with init_start_ts, warmup_done_ts,
|
|
first_frame_ts, warmup_ms, first_frame_after_init_ms,
|
|
first_frame_after_warmup_ms. Lets external clients answer
|
|
'did the warmup block the first frame?'.
|
|
[C: tests/test_api_hooks_warmup.py:test_live_startup_timeline_endpoint]
|
|
"""
|
|
return self._make_request('GET', '/api/startup_timeline') or {}
|
|
|
|
#endregion: Diagnostics
|
|
|
|
#region: Project
|
|
|
|
def get_project(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves the current project state.
|
|
[C: simulation/sim_context.py:ContextSimulation.run, tests/test_api_hook_client.py:test_get_project_success, tests/test_gui_context_presets.py:test_gui_context_preset_save_load, tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_hooks.py:test_live_hook_server_responses, tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
return self._make_request('GET', '/api/project') or {}
|
|
|
|
def get_project_switch_status(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the project switch status: {in_progress, path, error}.
|
|
- in_progress: True when a project switch is currently scheduled or running
|
|
- path: the path the switch is targeting (or the last completed switch path)
|
|
- error: string error message if the last switch failed; None on success
|
|
Used by tests to wait deterministically for a project switch to complete
|
|
instead of blind-polling the project state.
|
|
[C: tests/test_api_hooks_project_switch.py]
|
|
"""
|
|
result = self._make_request('GET', '/api/project_switch_status')
|
|
if not result or not isinstance(result, dict):
|
|
return {"in_progress": False, "path": None, "error": None}
|
|
return result
|
|
|
|
def wait_for_project_switch(self, expected_path: str = None, timeout: float = 30.0, poll_interval: float = 0.2) -> dict[str, Any]:
|
|
"""
|
|
Blocks until the project switch completes (or fails). Returns the
|
|
final status dict. If expected_path is provided, also waits until the
|
|
reported path matches it (or its basename matches).
|
|
Fails fast (within timeout seconds) with a clear error if:
|
|
- the controller reports an error during the switch
|
|
- the path doesn't reach expected_path within the timeout
|
|
- the controller is hung (in_progress stays True)
|
|
[C: tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
import os
|
|
start = time.time()
|
|
last_status = {"in_progress": True, "path": None, "error": None}
|
|
expected_basename = os.path.basename(expected_path) if expected_path else None
|
|
while time.time() - start < timeout:
|
|
last_status = self.get_project_switch_status()
|
|
if last_status.get("error"):
|
|
return last_status
|
|
if not last_status.get("in_progress"):
|
|
current_path = last_status.get("path") or ""
|
|
current_basename = os.path.basename(str(current_path))
|
|
path_matches = (expected_path is None
|
|
or current_path == expected_path
|
|
or (expected_basename and current_basename == expected_basename))
|
|
if path_matches:
|
|
return last_status
|
|
time.sleep(poll_interval)
|
|
last_status["timeout"] = True
|
|
return last_status
|
|
|
|
def get_io_pool_status(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the controller's io_pool status: {idle, inflight}.
|
|
- idle: True if no jobs are currently in-flight (running or queued)
|
|
- inflight: integer count of jobs submitted via submit_io
|
|
Used by tests to wait for the pool to drain before submitting work
|
|
when sharing a live_gui session with prior tests.
|
|
[C: tests/test_api_hook_client.py:test_get_io_pool_status_*]
|
|
"""
|
|
result = self._make_request('GET', '/api/io_pool_status')
|
|
if not result or not isinstance(result, dict):
|
|
return {"idle": True, "inflight": 0}
|
|
return result
|
|
|
|
def get_gui_health(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the controller's GUI health: {healthy, degraded_reason,
|
|
last_assert, io_pool_alive}. Tests should call this before starting
|
|
work and skip / fail fast if `healthy` is False.
|
|
- healthy: True if the GUI main loop is running normally
|
|
- degraded_reason: human-readable description of the failure (if any)
|
|
- last_assert: full traceback of the last ImGui scope mismatch
|
|
- io_pool_alive: True if submit_io is currently functional
|
|
[C: tests/test_api_hook_client_gui_health.py:test_get_gui_health_*]
|
|
"""
|
|
result = self._make_request('GET', '/api/gui_health')
|
|
if not result or not isinstance(result, dict):
|
|
return {
|
|
"healthy": True,
|
|
"degraded_reason": None,
|
|
"last_assert": None,
|
|
"io_pool_alive": True,
|
|
}
|
|
return result
|
|
|
|
def wait_io_pool_idle(self, timeout: float = 60.0, poll_interval: float = 0.2) -> bool:
|
|
"""
|
|
Blocks until the controller's io_pool reports idle=True or timeout.
|
|
Returns True on idle, False on timeout. Use this to ensure prior
|
|
tests' background work has completed before submitting new work.
|
|
[C: tests/test_live_workflow.py:test_full_live_workflow]
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
status = self.get_io_pool_status()
|
|
if status.get("idle"):
|
|
return True
|
|
time.sleep(poll_interval)
|
|
return False
|
|
|
|
def post_project(self, project_data: dict) -> dict[str, Any]:
|
|
return last_status
|
|
|
|
def post_project(self, project_data: dict) -> dict[str, Any]:
|
|
"""
|
|
Updates the current project configuration.
|
|
[C: simulation/sim_context.py:ContextSimulation.run]
|
|
"""
|
|
return self._make_request('POST', '/api/project', data=project_data) or {}
|
|
|
|
#endregion: Project
|
|
|
|
#region: Context
|
|
|
|
def inject_context(self, data: dict) -> dict:
|
|
"""
|
|
Injects custom file context into the application.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation]
|
|
"""
|
|
return self._make_request('POST', '/api/context/inject', data=data) or {}
|
|
|
|
def get_context_state(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves the current file and screenshot context state.
|
|
[C: tests/test_gui_context_presets.py:test_gui_context_preset_save_load]
|
|
"""
|
|
return self._make_request('GET', '/api/context/state') or {}
|
|
|
|
#endregion: Context
|
|
|
|
#region: Discussion
|
|
|
|
def get_session(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves the current discussion session history.
|
|
[C: simulation/ping_pong.py:main, simulation/sim_context.py:ContextSimulation.run, simulation/sim_execution.py:ExecutionSimulation.run, simulation/sim_tools.py:ToolsSimulation.run, simulation/workflow_sim.py:WorkflowSimulator.run_discussion_turn_async, simulation/workflow_sim.py:WorkflowSimulator.wait_for_ai_response, tests/test_api_hook_client.py:test_get_session_success, tests/test_gui_stress_performance.py:test_comms_volume_stress_performance, tests/test_live_workflow.py:test_full_live_workflow, tests/test_rag_phase4_final_verify.py:test_phase4_final_verify, tests/test_rag_phase4_stress.py:test_rag_large_codebase_verification_sim]
|
|
"""
|
|
return self._make_request('GET', '/api/session') or {}
|
|
|
|
def reset_session(self) -> None:
|
|
"""
|
|
Resets the current session via button click.
|
|
[C: src/app_controller.py:AppController._handle_reset_session, src/app_controller.py:AppController.current_model, src/app_controller.py:AppController.current_provider, src/app_controller.py:AppController.init_state, src/gui_2.py:App._render_provider_panel, src/gui_2.py:App._show_menus, src/multi_agent_conductor.py:run_worker_lifecycle, tests/conftest.py:live_gui, tests/conftest.py:reset_ai_client, tests/test_ai_cache_tracking.py:test_gemini_cache_tracking, tests/test_ai_client_cli.py:test_ai_client_send_gemini_cli, tests/test_api_events.py:test_send_emits_events_proper, tests/test_api_events.py:test_send_emits_tool_events, tests/test_deepseek_provider.py:test_deepseek_payload_verification, tests/test_deepseek_provider.py:test_deepseek_reasoner_payload_verification, tests/test_gemini_cli_integration.py:test_gemini_cli_full_integration, tests/test_gemini_cli_integration.py:test_gemini_cli_rejection_and_history, tests/test_gemini_metrics.py:test_get_gemini_cache_stats_with_mock_client, tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_mma_agent_focus_phase1.py:test_append_comms_has_source_tier_key, tests/test_mma_agent_focus_phase1.py:test_append_comms_source_tier_none_when_unset, tests/test_mma_agent_focus_phase1.py:test_append_comms_source_tier_set_when_current_tier_set, tests/test_mma_agent_focus_phase1.py:test_append_comms_source_tier_tier2, tests/test_session_logger_reset.py:test_reset_session, tests/test_token_usage.py:test_token_usage_tracking]
|
|
"""
|
|
self.click("btn_reset")
|
|
|
|
#endregion: Discussion
|
|
|
|
#region: Analytics
|
|
|
|
def get_financial_metrics(self) -> dict[str, Any]:
|
|
"""Retrieves token usage and estimated financial cost metrics."""
|
|
return self._make_request('GET', '/api/metrics/financial') or {}
|
|
|
|
def get_system_telemetry(self) -> dict[str, Any]:
|
|
"""Retrieves system-level telemetry including thread status and event queue size."""
|
|
return self._make_request('GET', '/api/system/telemetry') or {}
|
|
|
|
#endregion: Analytics
|
|
|
|
#region: MMA
|
|
|
|
def get_node_status(self, node_id: str) -> dict[str, Any]:
|
|
"""
|
|
Retrieves status for a specific node in the MMA DAG.
|
|
[C: tests/test_api_hook_client.py:test_get_node_status]
|
|
"""
|
|
return self._make_request('GET', f'/api/mma/node/{node_id}') or {}
|
|
|
|
def get_mma_status(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves the dedicated MMA engine status.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_live_workflow.py:test_full_live_workflow, tests/test_mma_concurrent_tracks_sim.py:_poll_mma_status, tests/test_mma_concurrent_tracks_sim.py:test_mma_concurrent_tracks_execution, tests/test_mma_concurrent_tracks_stress_sim.py:test_mma_concurrent_tracks_stress, tests/test_mma_step_mode_sim.py:_poll_mma_status, tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow, tests/test_visual_mma.py:test_visual_mma_components, tests/test_visual_orchestration.py:test_mma_epic_lifecycle, tests/test_visual_sim_gui_ux.py:test_gui_ux_event_routing, tests/test_visual_sim_mma_v2.py:_poll]
|
|
"""
|
|
return self._make_request('GET', '/api/gui/mma_status') or {}
|
|
|
|
def get_mma_workers(self) -> dict[str, Any]:
|
|
"""
|
|
Retrieves status for all active MMA workers.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation, tests/test_mma_concurrent_tracks_sim.py:test_mma_concurrent_tracks_execution, tests/test_mma_concurrent_tracks_stress_sim.py:_poll_mma_workers]
|
|
"""
|
|
return self._make_request('GET', '/api/mma/workers') or {}
|
|
|
|
def spawn_mma_worker(self, data: dict) -> dict:
|
|
"""
|
|
|
|
Spawns a new MMA worker with the provided configuration.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation]
|
|
"""
|
|
return self._make_request('POST', '/api/mma/workers/spawn', data=data) or {}
|
|
|
|
def kill_mma_worker(self, worker_id: str) -> dict:
|
|
"""Kills an active MMA worker by its ID."""
|
|
return self._make_request('POST', '/api/mma/workers/kill', data={"worker_id": worker_id}) or {}
|
|
|
|
def pause_mma_pipeline(self) -> dict:
|
|
"""
|
|
Pauses the MMA execution pipeline.
|
|
[C: tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow]
|
|
"""
|
|
return self._make_request('POST', '/api/mma/pipeline/pause') or {}
|
|
|
|
def resume_mma_pipeline(self) -> dict:
|
|
"""Resumes the MMA execution pipeline."""
|
|
return self._make_request('POST', '/api/mma/pipeline/resume') or {}
|
|
|
|
def mutate_mma_dag(self, data: dict) -> dict:
|
|
"""
|
|
Mutates the MMA DAG (Directed Acyclic Graph) structure.
|
|
[C: tests/test_headless_simulation.py:test_mma_track_lifecycle_simulation]
|
|
"""
|
|
return self._make_request('POST', '/api/mma/dag/mutate', data=data) or {}
|
|
|
|
def approve_mma_ticket(self, ticket_id: str) -> dict:
|
|
"""
|
|
Manually approves a specific ticket for execution in Step Mode.
|
|
[C: tests/test_mma_step_mode_sim.py:test_mma_step_mode_approval_flow]
|
|
"""
|
|
return self._make_request('POST', '/api/mma/ticket/approve', data={"ticket_id": ticket_id}) or {}
|
|
|
|
#endregion: MMA
|