Merge pull request #146 from SentienceAPI/assert_v2

Smart assert v1 + v2

Author: rcholic

.github/workflows/test.yml

@@ -299,17 +299,17 @@ jobs:
             sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
 
         print("=== Final Pre-Test Verification ===")
-        
+
         # First, verify the source file directly
         source_file = 'sentience/agent_runtime.py'
         print(f"=== Checking source file: {source_file} ===")
         if not os.path.exists(source_file):
             print(f"ERROR: Source file {source_file} not found!")
             sys.exit(1)
-        
+
         with open(source_file, 'r', encoding='utf-8') as f:
             source_content = f.read()
-        
+
         # Check if the bug exists and try to fix it one more time (in case auto-fix didn't run)
         if 'self.assertTrue(' in source_content:
             print('WARNING: Found self.assertTrue( in source file. Attempting to fix...')
@@ -332,11 +332,11 @@ jobs:
             print('OK: Source file uses self.assert_( correctly')
         else:
             print('WARNING: Could not find assert_ method in source file')
-        
+
         # Now check the installed package
         print("\n=== Checking installed package ===")
         import sentience.agent_runtime
-        
+
         # Verify it's using local source (editable install)
         import sentience
         pkg_path = os.path.abspath(sentience.__file__)
@@ -348,7 +348,7 @@ jobs:
             print(f'  This might be using PyPI package instead of local source!')
         else:
             print(f'OK: Package is from local source: {pkg_path}')
-        
+
         src = inspect.getsource(sentience.agent_runtime.AgentRuntime.assert_done)
 
         print("assert_done method source:")

README.md

@@ -31,13 +31,20 @@ Use `AgentRuntime` to add Jest-style assertions to your agent loops. Verify brow
 ```python
 import asyncio
 from sentience import AsyncSentienceBrowser, AgentRuntime
-from sentience.verification import url_contains, exists, all_of
+from sentience.verification import (
+    url_contains,
+    exists,
+    all_of,
+    is_enabled,
+    is_checked,
+    value_equals,
+)
 from sentience.tracing import Tracer, JsonlTraceSink
 
 async def main():
     # Create tracer
     tracer = Tracer(run_id="my-run", sink=JsonlTraceSink("trace.jsonl"))
-    
+
     # Create browser and runtime
     async with AsyncSentienceBrowser() as browser:
         page = await browser.new_page()
@@ -46,30 +53,43 @@ async def main():
             page=page,
             tracer=tracer
         )
-        
+
         # Navigate and take snapshot
         await page.goto("https://example.com")
         runtime.begin_step("Verify page loaded")
         await runtime.snapshot()
-        
-        # Run assertions (Jest-style)
+
+        # v1: deterministic assertions (Jest-style)
         runtime.assert_(url_contains("example.com"), label="on_correct_domain")
         runtime.assert_(exists("role=heading"), label="has_heading")
         runtime.assert_(all_of([
             exists("role=button"),
             exists("role=link")
         ]), label="has_interactive_elements")
-        
+
+        # v1: state-aware assertions (when Gateway refinement is enabled)
+        runtime.assert_(is_enabled("role=button"), label="button_enabled")
+        runtime.assert_(is_checked("role=checkbox name~'subscribe'"), label="subscribe_checked_if_present")
+        runtime.assert_(value_equals("role=textbox name~'email'", "user@example.com"), label="email_value_if_present")
+
+        # v2: retry loop with snapshot confidence gating + exhaustion
+        ok = await runtime.check(
+            exists("role=heading"),
+            label="heading_eventually_visible",
+            required=True,
+        ).eventually(timeout_s=10.0, poll_s=0.25, min_confidence=0.7, max_snapshot_attempts=3)
+        print("eventually() result:", ok)
+
         # Check task completion
         if runtime.assert_done(exists("text~'Example'"), label="task_complete"):
             print("✅ Task completed!")
-        
+
         print(f"Task done: {runtime.is_task_done}")
 
 asyncio.run(main())
 ```
 
-**See example:** [`examples/agent_runtime_verification.py`](examples/agent_runtime_verification.py)
+**See examples:** [`examples/asserts/`](examples/asserts/)
 
 ## 🚀 Quick Start: Choose Your Abstraction Level
 
@@ -183,56 +203,35 @@ scroll_to(browser, button.id, behavior='instant', block='start')
 ---
 
 <details>
-<summary><h2>💼 Real-World Example: Amazon Shopping Bot</h2></summary>
+<summary><h2>💼 Real-World Example: Assertion-driven navigation</h2></summary>
 
-This example demonstrates navigating Amazon, finding products, and adding items to cart:
+This example shows how to use **assertions + `.eventually()`** to make an agent loop resilient:
 
 ```python
-from sentience import SentienceBrowser, snapshot, find, click
-import time
-
-with SentienceBrowser(headless=False) as browser:
-    # Navigate to Amazon Best Sellers
-    browser.goto("https://www.amazon.com/gp/bestsellers/", wait_until="domcontentloaded")
-    time.sleep(2)  # Wait for dynamic content
-
-    # Take snapshot and find products
-    snap = snapshot(browser)
-    print(f"Found {len(snap.elements)} elements")
-
-    # Find first product in viewport using spatial filtering
-    products = [
-        el for el in snap.elements
-        if el.role == "link"
-        and el.visual_cues.is_clickable
-        and el.in_viewport
-        and not el.is_occluded
-        and el.bbox.y < 600  # First row
-    ]
-
-    if products:
-        # Sort by position (left to right, top to bottom)
-        products.sort(key=lambda e: (e.bbox.y, e.bbox.x))
-        first_product = products[0]
+import asyncio
+import os
+from sentience import AsyncSentienceBrowser, AgentRuntime
+from sentience.tracing import Tracer, JsonlTraceSink
+from sentience.verification import url_contains, exists
 
-        print(f"Clicking: {first_product.text}")
-        result = click(browser, first_product.id)
+async def main():
+    tracer = Tracer(run_id="verified-run", sink=JsonlTraceSink("trace_verified.jsonl"))
+    async with AsyncSentienceBrowser(headless=True) as browser:
+        page = await browser.new_page()
+        runtime = await AgentRuntime.from_sentience_browser(browser=browser, page=page, tracer=tracer)
+        runtime.sentience_api_key = os.getenv("SENTIENCE_API_KEY")  # optional, enables Gateway diagnostics
 
-        # Wait for product page
-        browser.page.wait_for_load_state("networkidle")
-        time.sleep(2)
+        await page.goto("https://example.com")
+        runtime.begin_step("Verify we're on the right page")
 
-        # Find and click "Add to Cart" button
-        product_snap = snapshot(browser)
-        add_to_cart = find(product_snap, "role=button text~'add to cart'")
+        await runtime.check(url_contains("example.com"), label="on_domain", required=True).eventually(
+            timeout_s=10.0, poll_s=0.25, min_confidence=0.7, max_snapshot_attempts=3
+        )
+        runtime.assert_(exists("role=heading"), label="heading_present")
 
-        if add_to_cart:
-            cart_result = click(browser, add_to_cart.id)
-            print(f"Added to cart: {cart_result.success}")
+asyncio.run(main())
 ```
 
-**📖 See the complete tutorial:** [Amazon Shopping Guide](../docs/AMAZON_SHOPPING_GUIDE.md)
-
 </details>
 
 ---

examples/asserts/README.md

@@ -0,0 +1,15 @@
+# Assertions examples (v1 + v2)
+
+These examples focus on **AgentRuntime assertions**:
+
+- **v1**: deterministic, state-aware assertions (enabled/checked/value/expanded) + failure intelligence
+- **v2**: `.eventually()` retry loops with `min_confidence` gating + snapshot exhaustion, plus optional Python vision fallback
+
+Run examples:
+
+```bash
+cd sdk-python
+python examples/asserts/v1_state_assertions.py
+python examples/asserts/v2_eventually_min_confidence.py
+python examples/asserts/v2_vision_fallback.py
+```

examples/asserts/eventually_min_confidence.py

@@ -0,0 +1,50 @@
+"""
+v2: `.check(...).eventually(...)` with snapshot confidence gating + exhaustion.
+
+This example shows:
+- retry loop semantics
+- `min_confidence` gating (snapshot_low_confidence -> snapshot_exhausted)
+- structured assertion records in traces
+"""
+
+import asyncio
+import os
+
+from sentience import AgentRuntime, AsyncSentienceBrowser
+from sentience.tracing import JsonlTraceSink, Tracer
+from sentience.verification import exists
+
+
+async def main() -> None:
+    tracer = Tracer(run_id="asserts-v2", sink=JsonlTraceSink("trace_asserts_v2.jsonl"))
+    sentience_api_key = os.getenv("SENTIENCE_API_KEY")
+
+    async with AsyncSentienceBrowser(headless=True) as browser:
+        page = await browser.new_page()
+        runtime = await AgentRuntime.from_sentience_browser(
+            browser=browser, page=page, tracer=tracer
+        )
+        if sentience_api_key:
+            runtime.sentience_api_key = sentience_api_key
+
+        await page.goto("https://example.com")
+        runtime.begin_step("Assert v2 eventually")
+
+        ok = await runtime.check(
+            exists("role=heading"),
+            label="heading_eventually_visible",
+            required=True,
+        ).eventually(
+            timeout_s=10.0,
+            poll_s=0.25,
+            # If the Gateway reports snapshot.diagnostics.confidence, gate on it:
+            min_confidence=0.7,
+            max_snapshot_attempts=3,
+        )
+
+        print("eventually() result:", ok)
+        print("Final assertion:", runtime.get_assertions_for_step_end()["assertions"])
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/asserts/state_assertions.py

@@ -0,0 +1,68 @@
+"""
+v1: State-aware assertions with AgentRuntime.
+
+This example is meant to be run with a Pro/Enterprise API key so the Gateway
+can refine raw elements into SmartElements with state fields (enabled/checked/value/etc).
+
+Env vars:
+  - SENTIENCE_API_KEY (optional but recommended for v1 state assertions)
+"""
+
+import asyncio
+import os
+
+from sentience import AgentRuntime, AsyncSentienceBrowser
+from sentience.tracing import JsonlTraceSink, Tracer
+from sentience.verification import (
+    exists,
+    is_checked,
+    is_disabled,
+    is_enabled,
+    is_expanded,
+    value_contains,
+)
+
+
+async def main() -> None:
+    tracer = Tracer(run_id="asserts-v1", sink=JsonlTraceSink("trace_asserts_v1.jsonl"))
+
+    sentience_api_key = os.getenv("SENTIENCE_API_KEY")
+
+    async with AsyncSentienceBrowser(headless=True) as browser:
+        page = await browser.new_page()
+        runtime = await AgentRuntime.from_sentience_browser(
+            browser=browser, page=page, tracer=tracer
+        )
+
+        # If you have a Pro/Enterprise key, set it on the runtime so snapshots use the Gateway.
+        # (This improves selector quality and unlocks state-aware fields for assertions.)
+        if sentience_api_key:
+            runtime.sentience_api_key = sentience_api_key
+
+        await page.goto("https://example.com")
+        runtime.begin_step("Assert v1 state")
+        await runtime.snapshot()
+
+        # v1: state-aware assertions (examples)
+        runtime.assert_(exists("role=heading"), label="has_heading")
+        runtime.assert_(is_enabled("role=link"), label="some_link_enabled")
+        runtime.assert_(
+            is_disabled("role=button text~'continue'"), label="continue_disabled_if_present"
+        )
+        runtime.assert_(
+            is_checked("role=checkbox name~'subscribe'"), label="subscribe_checked_if_present"
+        )
+        runtime.assert_(is_expanded("role=button name~'more'"), label="more_is_expanded_if_present")
+        runtime.assert_(
+            value_contains("role=textbox name~'email'", "@"), label="email_has_at_if_present"
+        )
+
+        # Failure intelligence: if something fails you’ll see:
+        # - details.reason_code
+        # - details.nearest_matches (suggestions)
+
+        print("Assertions recorded:", runtime.get_assertions_for_step_end()["assertions"])
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/asserts/vision_fallback.py

@@ -0,0 +1,58 @@
+"""
+v2 (Python-only): vision fallback after snapshot exhaustion.
+
+When `min_confidence` gating keeps failing (snapshot_exhausted), you can pass a
+vision-capable LLMProvider to `eventually()` and ask it for a strict YES/NO
+verification using a screenshot.
+
+Env vars:
+  - OPENAI_API_KEY (if using OpenAIProvider)
+  - SENTIENCE_API_KEY (optional, recommended so diagnostics/confidence is present)
+"""
+
+import asyncio
+import os
+
+from sentience import AgentRuntime, AsyncSentienceBrowser
+from sentience.llm_provider import OpenAIProvider
+from sentience.tracing import JsonlTraceSink, Tracer
+from sentience.verification import exists
+
+
+async def main() -> None:
+    tracer = Tracer(
+        run_id="asserts-v2-vision", sink=JsonlTraceSink("trace_asserts_v2_vision.jsonl")
+    )
+    sentience_api_key = os.getenv("SENTIENCE_API_KEY")
+
+    # Any provider implementing supports_vision() + generate_with_image() works.
+    vision = OpenAIProvider(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+    async with AsyncSentienceBrowser(headless=True) as browser:
+        page = await browser.new_page()
+        runtime = await AgentRuntime.from_sentience_browser(
+            browser=browser, page=page, tracer=tracer
+        )
+        if sentience_api_key:
+            runtime.sentience_api_key = sentience_api_key
+
+        await page.goto("https://example.com")
+        runtime.begin_step("Assert v2 vision fallback")
+
+        ok = await runtime.check(
+            exists("text~'Example Domain'"), label="example_domain_text"
+        ).eventually(
+            timeout_s=10.0,
+            poll_s=0.25,
+            min_confidence=0.7,
+            max_snapshot_attempts=2,
+            vision_provider=vision,
+            vision_system_prompt="You are a strict visual verifier. Answer only YES or NO.",
+            vision_user_prompt="In the screenshot, is the phrase 'Example Domain' visible? Answer YES or NO.",
+        )
+
+        print("eventually() w/ vision result:", ok)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())