fix(navigation): stabilize GramPilot autonomous grid navigation and Qdrant persistence
- Hardened `NavigationKnowledge` and `TelepathicEngine` to prevent premature blacklisting on high-latency grid taps by returning `None` (inconclusive) when grid markers are still present. - Updated `_execute_action` in `goap.py` to respect inconclusive verification states and avoid polluting the blacklist. - Refined `Adaptive Snap` in `timing.py` to detect `ProgressBar` loading spinners, extending timeouts for slow connections. - Prevented brittle Adaptive Snap `wobble` routines from firing while the bot is still trapped on the Explore Grid. - Stabilized Qdrant collections by converting destructive delete-only wipes into safe `wipe_collection` routines that instantly recreate the index. - Maintained 100% TDD pass-rate by aligning E2E grid navigation tests with the new inconclusive states.
This commit is contained in:
42
tests/integration/test_llm_edge_inference.py
Normal file
42
tests/integration/test_llm_edge_inference.py
Normal file
@@ -0,0 +1,42 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from GramAddict.core.llm_provider import query_telepathic_llm
|
||||
|
||||
def test_query_telepathic_llm_already_local():
|
||||
# If the provided URL is local, it should NOT switch to fallback model
|
||||
with patch("GramAddict.core.llm_provider.query_llm", return_value={"response": "test"}) as mock_query:
|
||||
query_telepathic_llm(
|
||||
model="llama3.2-vision",
|
||||
url="http://localhost:11434/api/generate",
|
||||
system_prompt="sys",
|
||||
user_prompt="user",
|
||||
use_local_edge=True
|
||||
)
|
||||
mock_query.assert_called_once()
|
||||
args, kwargs = mock_query.call_args
|
||||
assert kwargs["model"] == "llama3.2-vision"
|
||||
assert kwargs["url"] == "http://localhost:11434/api/generate"
|
||||
|
||||
def test_query_telepathic_llm_remote_with_local_edge():
|
||||
# If the provided URL is remote, it SHOULD switch to fallback model when edge=True
|
||||
|
||||
class MockArgs:
|
||||
ai_fallback_model = "llama3.2:1b"
|
||||
ai_fallback_url = "http://localhost:11434/api/generate"
|
||||
|
||||
class MockConfig:
|
||||
args = MockArgs()
|
||||
|
||||
with patch("GramAddict.core.llm_provider.query_llm", return_value={"response": "test"}) as mock_query:
|
||||
with patch("GramAddict.core.config.Config", return_value=MockConfig()):
|
||||
query_telepathic_llm(
|
||||
model="gpt-4o",
|
||||
url="https://api.openai.com/v1/chat/completions",
|
||||
system_prompt="sys",
|
||||
user_prompt="user",
|
||||
use_local_edge=True
|
||||
)
|
||||
mock_query.assert_called_once()
|
||||
args, kwargs = mock_query.call_args
|
||||
assert kwargs["model"] == "llama3.2:1b"
|
||||
assert kwargs["url"] == "http://localhost:11434/api/generate"
|
||||
49
tests/integration/test_qdrant_wipe.py
Normal file
49
tests/integration/test_qdrant_wipe.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from GramAddict.core.qdrant_memory import QdrantBase
|
||||
from GramAddict.core.telepathic_engine import TelepathicEngine
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_qdrant():
|
||||
with patch('GramAddict.core.qdrant_memory.QdrantClient') as mq:
|
||||
yield mq
|
||||
|
||||
def test_qdrant_wipe_recreates_collection(mock_qdrant):
|
||||
"""
|
||||
Tests that calling wipe_collection() on QdrantBase successfully calls
|
||||
delete_collection AND create_collection to prevent 404 errors.
|
||||
"""
|
||||
mock_client = MagicMock()
|
||||
mock_qdrant.return_value = mock_client
|
||||
|
||||
# Missing collection creation during init
|
||||
mock_client.collection_exists.return_value = False
|
||||
base = QdrantBase("test_collection", vector_size=4)
|
||||
|
||||
assert mock_client.create_collection.call_count == 1
|
||||
|
||||
# Now call wipe_collection
|
||||
base.wipe_collection()
|
||||
|
||||
mock_client.delete_collection.assert_called_with("test_collection")
|
||||
# create_collection should now have been called a 2nd time
|
||||
assert mock_client.create_collection.call_count == 2
|
||||
|
||||
def test_telepathic_engine_wipe_uses_wipe_collection(mock_qdrant):
|
||||
"""
|
||||
Tests that TelepathicEngine.wipe() uses the safe wipe_collection method.
|
||||
"""
|
||||
mock_client = MagicMock()
|
||||
mock_qdrant.return_value = mock_client
|
||||
|
||||
engine = TelepathicEngine()
|
||||
|
||||
# Spy on the wipe_collection method
|
||||
with patch.object(engine.embedding_helper, 'wipe_collection') as mock_emb_wipe, \
|
||||
patch.object(engine.ui_memory, 'wipe_collection') as mock_ui_wipe:
|
||||
|
||||
engine.wipe()
|
||||
|
||||
mock_emb_wipe.assert_called_once()
|
||||
mock_ui_wipe.assert_called_once()
|
||||
28
tests/integration/test_situational_awareness.py
Normal file
28
tests/integration/test_situational_awareness.py
Normal file
@@ -0,0 +1,28 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
|
||||
|
||||
@pytest.fixture
|
||||
def sae():
|
||||
device = MagicMock()
|
||||
device.app_id = "com.instagram.android"
|
||||
device.deviceV2.info = {"screenOn": True}
|
||||
return SituationalAwarenessEngine(device)
|
||||
|
||||
def test_perceive_normal_with_unknown_keyboard(sae):
|
||||
# XML contains Instagram and some unknown keyboard
|
||||
xml = """
|
||||
<hierarchy>
|
||||
<node package="com.instagram.android" resource-id="com.instagram.android:id/button" />
|
||||
<node package="com.unknown.keyboard" resource-id="com.unknown.keyboard:id/key" />
|
||||
</hierarchy>
|
||||
"""
|
||||
|
||||
# We shouldn't call LLM for foreign app
|
||||
with patch('GramAddict.core.llm_provider.query_telepathic_llm') as mock_llm:
|
||||
# Let's mock ScreenMemoryDB to return NORMAL
|
||||
with patch('GramAddict.core.qdrant_memory.ScreenMemoryDB.get_screen_type', return_value="NORMAL"):
|
||||
res = sae.perceive(xml)
|
||||
assert res == SituationType.NORMAL
|
||||
# The LLM for foreign app should NOT have been called.
|
||||
mock_llm.assert_not_called()
|
||||
@@ -166,10 +166,10 @@ class TestVerifySuccessExploreGrid:
|
||||
just didn't register; it doesn't mean the mapping is wrong.
|
||||
"""
|
||||
|
||||
def test_verify_success_returns_false_when_still_on_grid(self):
|
||||
def test_verify_success_returns_none_when_still_on_grid(self):
|
||||
"""
|
||||
If we tapped a grid item but the screen still shows the explore grid
|
||||
(no feed markers), verify_success must return False.
|
||||
(no feed markers), verify_success must return None (Inconclusive).
|
||||
"""
|
||||
engine = TelepathicEngine()
|
||||
|
||||
@@ -192,7 +192,7 @@ class TestVerifySuccessExploreGrid:
|
||||
"""
|
||||
|
||||
result = engine.verify_success("first image in explore grid", still_on_grid_xml)
|
||||
assert result is False, "verify_success should fail when still on explore grid"
|
||||
assert result is None, "verify_success should be inconclusive (None) when still on explore grid"
|
||||
|
||||
def test_verify_success_returns_true_when_post_opened(self):
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user