ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity
Files
508baa69b63cfc1bf1edd07458408111c1ddd55c
manual_slop/tests/test_visual_sim_mma_v2.py
T
ed 9cfbb980bd fix(mma_lifecycle): load + start + active_tickets sync for batched tests 
The test_visual_sim_mma_v2 failure in tier-3 batch context was caused
by state pollution from prior live_gui tests sharing the subprocess:

1. track state file missing for leftover track:
   `_cb_load_track_result` accessed `state.metadata.id` and
   `state.metadata.name`, but `EMPTY_TRACK_STATE` (returned when no
   state.toml exists for a track_id) had `metadata={}` (a dict, not a
   TrackMetadata object). That raised `'dict' object has no attribute
   'id'`. Fixed by normalizing metadata: dict -> TrackMetadata.from_dict,
   TrackMetadata stays, anything else -> TrackMetadata(id=track_id,
   name=track_id).

2. active_track and active_tickets never reached the App:
   `_cb_load_track_result` set `self.active_track` (controller) and
   `self.active_tickets = []` (via `_load_active_tickets`) but never
   mirrored to `self._app.active_track` / `self._app.active_tickets`.
   The /api/gui/mma_status endpoint reads `app.X` first via
   `_get_app_attr`, so it returned None / [] and the test's poll
   `at_id == track_id and bool(s.get('active_tickets'))` failed. Fixed
   by mirroring active_track / active_tickets / active_tier to the App.

3. leftover tracks list in batched run:
   Without btn_reset, `app.tracks` (App-side) accumulates tracks from
   earlier tests in the session. `_get_app_attr(app, 'tracks', [])`
   then returns stale leftovers, and the test's
   `target_track = next((... if 'hello_world' in t.get('title') else
   tracks_list[0]))` picks a leftover with no on-disk state file.
   Fixed in TWO places:
   (a) btn_reset now also clears `app.tracks = []` so each test starts
       clean if it calls btn_reset.
   (b) tests/test_visual_sim_mma_v2.py now calls `client.click('btn_reset')`
       at the start. The test was the only one in its tier that did NOT
       reset; with the live_gui subprocess shared across batched tests,
       that's the source of the state pollution.

Also reverted the `TrackState.metadata` default change (dict -> TrackMetadata)
because it broke TrackState() construction (TrackMetadata requires `id`/`name`).
The metadata normalization in `_cb_load_track_result` is sufficient and
preserves backward compatibility with on-disk state.toml files.

Verified: tests/test_visual_sim_mma_v2.py passes in isolation (58.36s)
and tier-3 batch passes when run standalone. Other tests in tier-3
have pre-existing render-loop contention flakes in batched xdist mode
that are unrelated to this fix.
2026-07-02 10:23:14 -04:00

219 lines
9.0 KiB
Python
Raw Blame History

import pytest
import time
import sys
import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
from src import api_hook_client
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _drain_approvals(client: api_hook_client.ApiHookClient, status: dict) -> None:
"""Auto-approve any pending approval gate found in status."""
if status.get('pending_mma_spawn_approval'):
print('[SIM] Approving pending spawn...')
client.click('btn_approve_spawn')
time.sleep(0.5)
elif status.get('pending_mma_step_approval'):
print('[SIM] Approving pending MMA step...')
client.click('btn_approve_mma_step')
time.sleep(0.5)
elif status.get('pending_tool_approval'):
print('[SIM] Approving pending tool...')
client.click('btn_approve_tool')
time.sleep(0.5)
elif status.get('pending_script_approval'):
print('[SIM] Approving pending PowerShell script...')
client.click('btn_approve_script')
time.sleep(0.5)
def _poll(client: api_hook_client.ApiHookClient, timeout: int, condition, label: str) -> tuple[bool, dict]:
"""Poll get_mma_status() until condition(status) is True or timeout."""
status = {}
for i in range(timeout):
status = client.get_mma_status() or {}
print(f"[SIM][{label}] t={i}s ai_status={status.get('ai_status')} "
f"mma={status.get('mma_status')} "
f"streams={list(status.get('mma_streams', {}).keys())}")
_drain_approvals(client, status)
if condition(status):
return True, status
time.sleep(1)
return False, status
# ---------------------------------------------------------------------------
# Test
# ---------------------------------------------------------------------------
@pytest.mark.integration
@pytest.mark.timeout(300)
def test_mma_complete_lifecycle(live_gui) -> None:
"""
End-to-end MMA lifecycle using real Gemini API (gemini-2.5-flash-lite).
Incorporates frame-sync sleeps and explicit state-transition waits per
simulation_hardening_20260301 spec (Issues 2 & 3).
"""
client = api_hook_client.ApiHookClient()
assert client.wait_for_server(timeout=15), "Hook server did not start"
# Clear any stale state from prior live_gui tests in this batched
# subprocess. Without btn_reset, leftover tracks (created by earlier
# tests like test_mma_concurrent_tracks_sim / test_visual_mma) sit in
# `app.tracks` and become `tracks_list[0]`. The test's `target_track =
# next(..., tracks_list[0])` then loads a leftover track whose on-disk
# state file does not exist, so `_cb_load_track_result` falls back to
# EMPTY_TRACK_STATE → `state.tasks == []` → `bool(active_tickets)` is
# False → Stage 6 polls fail. btn_reset clears `app.tracks` / `app.active_tickets`
# / `app.active_track` so `tracks_list` reflects only tracks accepted
# in this test's plan_epic + accept_tracks batch.
client.click("btn_reset")
time.sleep(2)
# ------------------------------------------------------------------
# Stage 1: Provider setup
# ------------------------------------------------------------------
client.set_value('current_provider', 'gemini_cli')
time.sleep(0.3)
client.set_value('gcli_path', f'"{sys.executable}" "{os.path.abspath("tests/mock_gemini_cli.py")}"')
time.sleep(0.3)
# Per Tier 1 investigation: do NOT change files_base_dir here and do NOT
# click btn_project_save. The previous version set files_base_dir to
# 'tests/artifacts/temp_workspace' and persisted it via btn_project_save,
# which polluted the session-scoped subprocess's ui_files_base_dir (and
# the workspace's manual_slop.toml) with a test-only path that persisted
# into subsequent live_gui tests (e.g. test_rag_phase4_final_verify).
# The MMA lifecycle does not depend on a specific files_base_dir — the
# mock_gemini_cli returns canned responses regardless.
time.sleep(0.3)
# ------------------------------------------------------------------
# Stage 2: Start epic planning
# ------------------------------------------------------------------
# Keep prompt short and simple so the model returns minimal JSON
client.set_value('mma_epic_input',
'PATH: Epic Initialization')
time.sleep(0.3)
client.click('btn_mma_plan_epic')
time.sleep(0.5) # frame-sync after click
# ------------------------------------------------------------------
# Stage 3: Wait for proposed_tracks to appear (Tier 1 call)
# ------------------------------------------------------------------
ok, status = _poll(client, timeout=120, label="wait-proposed-tracks",
condition=lambda s: bool(s.get('proposed_tracks')))
assert ok, (
f"No proposed_tracks after 120s. "
f"ai_status={status.get('ai_status')} "
f"mma_streams={list(status.get('mma_streams', {}).keys())}"
)
n_proposed = len(status['proposed_tracks'])
print(f"[SIM] Got {n_proposed} proposed track(s): "
f"{[t.get('title', t.get('id')) for t in status['proposed_tracks']]}")
# ------------------------------------------------------------------
# Stage 4: Accept tracks (triggers Tier 2 calls + engine.run)
# ------------------------------------------------------------------
client.click('btn_mma_accept_tracks')
time.sleep(1.5) # frame-sync: let _cb_accept_tracks run one frame + bg thread start
# ------------------------------------------------------------------
# Stage 5: Wait for tracks to be written to filesystem + refreshed
# ------------------------------------------------------------------
ok, status = _poll(client, timeout=90, label="wait-tracks-populated",
condition=lambda s: bool(s.get('tracks')))
assert ok, (
f"No tracks appeared after 90s. "
f"ai_status={status.get('ai_status')}"
)
tracks_list = status['tracks']
print(f"[SIM] Tracks in project: {[t.get('title', t.get('id')) for t in tracks_list]}")
# ------------------------------------------------------------------
# Stage 6: Load first track, verify active_tickets populate
# ------------------------------------------------------------------
target_track = next((t for t in tracks_list if "hello_world" in t.get('title', '')), tracks_list[0])
track_id = target_track['id']
print(f"[SIM] Loading track: {track_id}")
client.click('btn_mma_load_track', user_data=track_id)
time.sleep(1.0) # frame-sync after load click
print(f"[SIM] Starting track: {track_id}")
client.click('btn_mma_start_track', user_data=track_id)
time.sleep(1.0) # frame-sync after start click
def _track_loaded(s):
at = s.get('active_track')
at_id = at.get('id') if isinstance(at, dict) else at
return at_id == track_id and bool(s.get('active_tickets'))
ok, status = _poll(client, timeout=60, label="wait-track-loaded",
condition=_track_loaded)
assert ok, (
f"Track {track_id} did not load with tickets after 60s. "
f"active_track={status.get('active_track')}"
)
print(f"[SIM] Track loaded with {len(status.get('active_tickets', []))} ticket(s).")
# ------------------------------------------------------------------
# Stage 7: Wait for engine to reach running/done
# ------------------------------------------------------------------
def _mma_active(s):
return s.get('mma_status') in ('running', 'done')
ok, status = _poll(client, timeout=120, label="wait-mma-running",
condition=_mma_active)
assert ok, (
f"MMA never reached running/done after 120s. "
f"mma_status={status.get('mma_status')}"
)
print(f"[SIM] MMA status: {status.get('mma_status')}")
# ------------------------------------------------------------------
# Stage 8: Verify Tier 3 output appears in mma_streams
# ------------------------------------------------------------------
def _tier3_in_streams(s):
streams = s.get('mma_streams', {})
tier3_keys = [k for k in streams if 'Tier 3' in k]
if not tier3_keys:
return False
return bool(streams[tier3_keys[0]].strip())
ok, status = _poll(client, timeout=120, label="wait-tier3-streams",
condition=_tier3_in_streams)
streams = status.get('mma_streams', {})
tier3_keys = [k for k in streams if 'Tier 3' in k]
assert ok, (
f"No non-empty Tier 3 output in mma_streams after 120s. "
f"streams keys={list(streams.keys())} "
f"mma_status={status.get('mma_status')}"
)
tier3_content = streams[tier3_keys[0]]
print(f"[SIM] Tier 3 output ({len(tier3_content)} chars): {tier3_content[:100]}...")
# ------------------------------------------------------------------
# Stage 9: Wait for mma_status == 'done' and mma_tier_usage Tier 3 non-zero
# ------------------------------------------------------------------
def _tier3_usage_nonzero(s):
usage = s.get('mma_tier_usage', {})
t3 = usage.get('Tier 3', {})
return t3.get('input', 0) > 0 or t3.get('output', 0) > 0
ok, status = _poll(client, timeout=30, label="wait-tier3-usage",
condition=_tier3_usage_nonzero)
# Non-blocking: if tier_usage isn't wired yet, just log and continue
tier_usage = status.get('mma_tier_usage', {})
print(f"[SIM] Tier usage: {tier_usage}")
if not ok:
print("[SIM] WARNING: mma_tier_usage Tier 3 still zero after 30s — may not be wired to hook API yet")
print("[SIM] MMA complete lifecycle simulation PASSED.")
View Git Blame Copy Permalink