ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sim): add pytest timeout(300) and tier_usage Stage 9 check

Browse Source 
Task 2.3: prevent infinite CI hangs with 300s hard timeout
Task 3.2: non-blocking Stage 9 logs mma_tier_usage after Tier 3 completes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Edward R. Gonzalez
2026-03-01 14:24:05 -05:00
parent 08734532ce
commit 63fa181192
2 changed files with 49 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
scripts/tasks/task_2_3_and_3_2.toml Normal file
View File

@@ -0,0 +1,31 @@
role = "tier3-worker"
docs = ["conductor/workflow.md", "tests/visual_sim_mma_v2.py"]
prompt = """
Make two additions to tests/visual_sim_mma_v2.py:
CHANGE 1 — Task 2.3: Add pytest timeout to prevent infinite CI hangs.
Add @pytest.mark.timeout(300) decorator to the test_mma_complete_lifecycle function.
Also add 'timeout' to the existing pytest.mark.integration decorator line (keep both marks).
CHANGE 2 — Task 3.2: Add tier_usage assertion after the existing Stage 8 check.
After the existing assertion on tier3_content, add a new polling stage:
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")
Add this before the final print("[SIM] MMA complete lifecycle simulation PASSED.") line.
Use exactly 1-space indentation for Python code.
"""

18
tests/visual_sim_mma_v2.py
View File

@@ -47,6 +47,7 @@ def _poll(client: ApiHookClient, timeout: int, condition, label: str) -> tuple[b
# ---------------------------------------------------------------------------
@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).
@@ -168,4 +169,21 @@ def test_mma_complete_lifecycle(live_gui) -> None:
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.")