Agent Tracing & Debugging

from predicate import PredicateBrowser, PredicateAgent
from predicate.llm_provider import OpenAIProvider
from predicate.tracing import Tracer, JsonlTraceSink
from predicate.agent_config import AgentConfig

# 1. Create a tracer with JSONL file sink
tracer = Tracer(
    run_id="shopping-bot-run-123",
    sink=JsonlTraceSink("trace.jsonl")
)

# 2. Configure agent behavior (optional)
config = AgentConfig(
    snapshot_limit=50,           # Max elements per snapshot
    temperature=0.0,             # LLM temperature
    max_retries=1,               # Retries on failure
    capture_screenshots=True,    # Include screenshots in traces
    screenshot_format="jpeg",    # jpeg or png
    screenshot_quality=80        # 1-100 for JPEG
)

# 3. Create agent with tracing enabled
browser = PredicateBrowser()
llm = OpenAIProvider(api_key="your-key", model="gpt-4o")
agent = PredicateAgent(browser, llm, tracer=tracer, config=config)

# 4. Use agent normally - all actions are automatically traced
with browser:
    browser.page.goto("https://amazon.com")
    agent.act("Click the search box")
    agent.act("Type 'magic mouse' into search")
    agent.act("Press Enter")

# Trace events are written to trace.jsonl

from predicate.agent_runtime import AgentRuntime

# Default: auto-emit is enabled (emit_trace=True)
snapshot = await runtime.snapshot()  # Automatically emits snapshot trace event

# The trace now includes screenshot_base64 for Studio visualization
# No additional code needed!

# Disable auto-emit for manual control
snapshot = await runtime.snapshot(emit_trace=False)

# Later, manually emit the snapshot when ready
tracer.emit_snapshot(
    snapshot=snapshot,
    step_id=runtime.step_id,
    step_index=runtime.step_index
)

from predicate.tracing import Tracer, JsonlTraceSink

tracer = Tracer(run_id="my-run", sink=JsonlTraceSink("trace.jsonl"))

# Emit snapshot event with screenshot for Studio visualization
tracer.emit_snapshot(
    snapshot=my_snapshot,           # Snapshot object with screenshot attribute
    step_id="step-uuid-123",        # Correlate with step (optional)
    step_index=1,                   # For Studio timeline ordering (optional)
    screenshot_format="jpeg"        # "jpeg" or "png" (default: "jpeg")
)

# Python: Pillow is optional but recommended for image manipulation
pip install Pillow

# Or install with the vision-local extras
pip install sentienceapi[vision-local]

from predicate.tracer_factory import create_tracer
import base64
from io import BytesIO

def redact_pii(screenshot_base64: str) -> str:
    """Redact PII by blurring sensitive regions."""
    try:
        from PIL import Image, ImageFilter
    except ImportError:
        # Pillow not installed - return original
        return screenshot_base64

    # Decode base64 to image
    image_data = base64.b64decode(screenshot_base64)
    img = Image.open(BytesIO(image_data))

    # Example: blur the entire image (replace with targeted redaction)
    blurred = img.filter(ImageFilter.GaussianBlur(radius=5))

    # Re-encode to base64
    output = BytesIO()
    blurred.save(output, format="JPEG", quality=80)
    return base64.b64encode(output.getvalue()).decode("utf-8")

tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="my-run",
    upload_trace=True,
    screenshot_processor=redact_pii  # Applied to all screenshots
)

from predicate.tracer_factory import create_tracer

def log_screenshot(screenshot_base64: str) -> str:
    """Log screenshot size without modifying it."""
    size_kb = len(screenshot_base64) * 3 / 4 / 1024  # Approximate decoded size
    print(f"Screenshot captured: ~{size_kb:.1f} KB")
    return screenshot_base64  # Return unchanged

tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="my-run",
    upload_trace=True,
    screenshot_processor=log_screenshot
)

from predicate.tracer_factory import create_tracer

# run_start is automatically emitted with metadata
tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="my-run-123",
    upload_trace=True,
    goal="Add headphones to cart",        # Displayed as trace name in Studio
    agent_type="PredicateAgent",          # Agent identifier
    llm_model="gpt-4o",                   # LLM model used
    start_url="https://amazon.com",       # Starting URL
)
# run_start event is automatically emitted here!

# To disable auto-emit (for manual control):
tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="my-run-123",
    auto_emit_run_start=False,  # Disable auto-emit
)
# Now you must manually emit:
tracer.emit_run_start("MyAgent", "gpt-4o", {"goal": "Custom goal"})

from predicate.agent_runtime import AgentRuntime

# step_start is automatically emitted when you begin a step
step_id = runtime.begin_step(
    goal="Click the search button",
    step_index=1,
    pre_url="https://example.com",  # Optional: URL before step
)
# step_start event is automatically emitted here!

# To disable auto-emit (for manual control):
step_id = runtime.begin_step(
    goal="Click the search button",
    step_index=1,
    emit_trace=False,  # Disable auto-emit
)
# Now you must manually emit:
tracer.emit_step_start(
    step_id=step_id,
    step_index=1,
    goal="Click the search button",
    attempt=0,
    pre_url="https://example.com"
)

# After all steps are complete, emit run_end
tracer.emit_run_end(
    steps=total_steps,           # Total number of steps executed
    status="success"             # "success", "failure", "partial", or "unknown"
)

# Then close the tracer to upload
tracer.close()

from predicate.tracer_factory import create_tracer
from predicate.agent_runtime import AgentRuntime

# 1. Create tracer - run_start is auto-emitted
tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="complete-example",
    upload_trace=True,
    goal="Search for products",
    agent_type="PredicateAgent",
    llm_model="gpt-4o",
    start_url="https://example.com",
)

# 2. Create runtime with tracer
runtime = AgentRuntime(page=page, tracer=tracer)

# 3. Begin step - step_start is auto-emitted
step_id = runtime.begin_step(goal="Click search", step_index=1)

# 4. Take snapshot - snapshot event is auto-emitted
snapshot = await runtime.snapshot()

# 5. Execute action and end step
# ... your action execution code ...
runtime.end_step(success=True)

# 6. Emit run_end (manual)
tracer.emit_run_end(steps=1, status="success")

# 7. Close tracer to upload
tracer.close()

from predicate.agent_config import AgentConfig

config = AgentConfig(
    # Snapshot settings
    snapshot_limit=50,              # Max elements to include (default: 50)

    # LLM settings
    temperature=0.0,                # LLM temperature 0.0-1.0 (default: 0.0)
    max_retries=1,                  # Retries on failure (default: 1)

    # Verification
    verify=True,                    # Verify action success (default: True)

    # Screenshot settings
    capture_screenshots=True,       # Capture screenshots (default: True)
    screenshot_format="jpeg",       # "jpeg" or "png" (default: "jpeg")
    screenshot_quality=80           # 1-100 for JPEG (default: 80)
)

agent = PredicateAgent(browser, llm, config=config)

from predicate import snapshot
from predicate.utils import compute_snapshot_digests

snap = snapshot(browser)

# Compute both strict and loose digests
digests = compute_snapshot_digests(snap.elements)

print(digests["strict"])  # sha256:abc123... (changes if text changes)
print(digests["loose"])   # sha256:def456... (only changes if layout changes)

# Use for loop detection
if current_digest == previous_digest:
    print("Agent is stuck in a loop!")

from predicate.formatting import format_snapshot_for_llm

snap = snapshot(browser)

# Format top 50 elements for LLM context
llm_context = format_snapshot_for_llm(snap, limit=50)

print(llm_context)
# Output:
# [1] <button> "Sign In" {PRIMARY,CLICKABLE} @ (100,50) (Imp:10)
# [2] <input> "Email address" {CLICKABLE} @ (100,100) (Imp:8)
# [3] <link> "Forgot password?" @ (150,140) (Imp:5)

from predicate import PredicateBrowser, PredicateAgent
from predicate.llm_provider import OpenAIProvider
from predicate.tracer_factory import create_tracer

# 1. Create tracer with automatic tier detection

tracer = create_tracer(
    api_key="sk_pro_xxxxx",  # Pro/Builder/Teams/Enterprise tier key
    run_id="shopping-bot-123",  # Gateway requires UUID format
    upload_trace=True,  # Set to True if you want cloud upload
    goal="Buy a laptop from Amazon",
    agent_type="Amazon Shopping Agent",
    llm_model="gpt-4o",
    start_url="https://www.amazon.com",
    screenshot_processor=None, # function for PII redaction
)

# 2. Create agent with tracer
browser = PredicateBrowser(api_key="sk_pro_xxxxx")
llm = OpenAIProvider(api_key="your_openai_key", model="gpt-4o")
agent = PredicateAgent(browser, llm, tracer=tracer)

# 3. Use agent normally - traces automatically uploaded
with browser:
    browser.page.goto("https://amazon.com")
    agent.act("Click the search box")
    agent.act("Type 'wireless mouse' into search")
    agent.act("Press Enter")

# 4. Upload to cloud (happens automatically on close)
tracer.close()  # Uploads trace + screenshots to cloud

from predicate.tracing import TraceSink

class CustomTraceSink(TraceSink):
    """Base interface for trace storage"""

    def emit(self, event_dict: dict) -> None:
        """Write a single trace event"""
        raise NotImplementedError

    def close(self) -> None:
        """Close the sink and flush any pending writes"""
        raise NotImplementedError

from predicate.tracing import TraceSink, Tracer
import psycopg2
import json

class DatabaseTraceSink(TraceSink):
    """Store traces in PostgreSQL database"""

    def __init__(self, connection_string: str):
        self.conn = psycopg2.connect(connection_string)
        self.cursor = self.conn.cursor()

        # Create traces table if it doesn't exist
        self.cursor.execute("""
            CREATE TABLE IF NOT EXISTS trace_events (
                id SERIAL PRIMARY KEY,
                run_id TEXT NOT NULL,
                seq INTEGER NOT NULL,
                type TEXT NOT NULL,
                timestamp TIMESTAMPTZ NOT NULL,
                data JSONB NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        self.conn.commit()

    def emit(self, event_dict: dict) -> None:
        """Insert trace event into database"""
        self.cursor.execute("""
            INSERT INTO trace_events (run_id, seq, type, timestamp, data)
            VALUES (%s, %s, %s, %s, %s)
        """, (
            event_dict["run_id"],
            event_dict["seq"],
            event_dict["type"],
            event_dict["ts"],
            json.dumps(event_dict["data"])
        ))
        self.conn.commit()

    def close(self) -> None:
        """Close database connection"""
        self.cursor.close()
        self.conn.close()

# Usage
tracer = Tracer(
    run_id="run-123",
    sink=DatabaseTraceSink("postgresql://user:pass@localhost/traces")
)
agent = PredicateAgent(browser, llm, tracer=tracer)

from predicate.tracing import TraceSink
import boto3
import json
from typing import List

class S3TraceSink(TraceSink):
    """Store traces in AWS S3"""

    def __init__(self, bucket: str, prefix: str = "traces/"):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.prefix = prefix
        self.events: List[dict] = []

    def emit(self, event_dict: dict) -> None:
        """Buffer trace events in memory"""
        self.events.append(event_dict)

    def close(self) -> None:
        """Upload all events to S3"""
        if not self.events:
            return

        run_id = self.events[0]["run_id"]
        key = f"{self.prefix}{run_id}.jsonl"

        # Convert events to JSONL
        jsonl_content = "\n".join(json.dumps(e) for e in self.events)

        # Upload to S3
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=jsonl_content.encode('utf-8'),
            ContentType='application/x-ndjson'
        )
        print(f"✅ Uploaded trace to s3://{self.bucket}/{key}")

# Usage
tracer = Tracer(
    run_id="run-123",
    sink=S3TraceSink(bucket="my-traces", prefix="production/")
)

from predicate.tracing import TraceSink, JsonlTraceSink
from typing import List

class MultiSink(TraceSink):
    """Write to multiple trace sinks simultaneously"""

    def __init__(self, sinks: List[TraceSink]):
        self.sinks = sinks

    def emit(self, event_dict: dict) -> None:
        """Emit to all sinks"""
        for sink in self.sinks:
            sink.emit(event_dict)

    def close(self) -> None:
        """Close all sinks"""
        for sink in self.sinks:
            sink.close()

# Usage: Write to both local file and database
tracer = Tracer(
    run_id="run-123",
    sink=MultiSink([
        JsonlTraceSink("trace.jsonl"),
        DatabaseTraceSink("postgresql://localhost/traces"),
        S3TraceSink("my-traces")
    ])
)

from typing import Protocol

class PredicateLogger(Protocol):
    """Protocol for optional logger interface."""

    def info(self, message: str) -> None:
        """Log info message."""
        ...

    def warning(self, message: str) -> None:
        """Log warning message."""
        ...

    def error(self, message: str) -> None:
        """Log error message."""
        ...

import logging
from predicate import create_tracer

# Use Python's built-in logging module
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Add handler to output to console
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('[%(levelname)s] %(message)s'))
logger.addHandler(handler)

# Create tracer with logger
tracer = create_tracer(
  api_key="sk_pro_xxxxx",
  run_id=f"amazon-shopping-{datetime.now().strftime('%Y%m%d-%H%M%S')}",
  logger=logger  # Pass standard Python logger (implements the protocol)
  upload_trace=True,
  goal="Buy a laptop from Amazon",
  agent_type="Amazon Shopping Agent",
  llm_model="gpt-4o",
  start_url="https://www.amazon.com",
  screenshot_processor=None, # function for PII redaction
)

# The logger will receive messages like:
# [INFO] Trace file size: 2.45 MB
# [INFO] Screenshot total: 0.00 MB
# [INFO] Trace completion reported to gateway
# [WARN] Failed to report trace completion: HTTP 500

from predicate import create_tracer

class CustomLogger:
    """Custom logger implementation"""

    def info(self, message: str) -> None:
        print(f"[INFO] {message}")
        # Send to monitoring service, file, etc.

    def warning(self, message: str) -> None:
        print(f"[WARN] {message}")
        # Alert team, log to error tracking

    def error(self, message: str) -> None:
        print(f"[ERROR] {message}")
        # Critical alert, page on-call engineer

custom_logger = CustomLogger()
tracer = create_tracer(
    api_key="sk_pro_xxxxx",  # Pro/Builder/Teams/Enterprise tier key
    run_id="shopping-bot-123",  # Gateway requires UUID format
    upload_trace=True,  # Set to True if you want cloud upload
    goal="Buy a laptop from Amazon",
    agent_type="Amazon Shopping Agent",
    llm_model="gpt-4o",
    start_url="https://www.amazon.com",
    screenshot_processor=None, # function for PII redaction
)

from predicate import create_tracer, PredicateBrowser, PredicateAgent
from predicate.llm import OpenAIProvider

# Run 1: Agent crashes mid-execution
tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="run-1",
    upload_trace=True,
    goal="Click button on example.com",
    agent_type="Example Agent",
    llm_model="gpt-4o",
    start_url="https://example.com",
    screenshot_processor=None, # function for PII redaction
)
browser = PredicateBrowser(api_key="sk_pro_xxxxx")
llm = OpenAIProvider(api_key="your_key", model="gpt-4o")
agent = PredicateAgent(browser, llm, tracer=tracer)

with browser:
    browser.page.goto("https://example.com")
    agent.act("Click button")  # Process crashes here - trace saved locally

# Run 2: SDK automatically recovers and uploads orphaned trace
tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="run-2",
    upload_trace=True,
    goal="Continue from previous run",
    agent_type="Example Agent",
    llm_model="gpt-4o",
    start_url="https://example.com",
    screenshot_processor=None, # function for PII redaction
)
# Prints: "⚠️  [Predicate] Found 1 un-uploaded trace(s) from previous runs"
# Prints: "✅ Uploaded orphaned trace: run-1"

browser = PredicateBrowser(api_key="sk_pro_xxxxx")
agent = PredicateAgent(browser, llm, tracer=tracer)
# Continue with run-2...

from predicate import create_tracer

tracer = create_tracer(
    api_key="sk_pro_xxxxx",
    run_id="run-123",
    upload_trace=True,
    goal="Example task",
    agent_type="Example Agent",
    llm_model="gpt-4o",
    start_url="https://example.com",
    screenshot_processor=None, # function for PII redaction
)

# ... agent execution ...

# Option 1: Blocking (default) - waits for upload to complete
tracer.close(blocking=True)  # Script pauses here until upload finishes
print("Upload complete!")

# Option 2: Non-blocking - returns immediately
tracer.close(blocking=False)  # Script continues immediately
print("Upload started in background!")
# Script can exit or continue with other work

def progress_callback(uploaded_bytes: int, total_bytes: int):
    percent = (uploaded_bytes / total_bytes) * 100
    print(f"Upload progress: {percent:.1f}% ({uploaded_bytes}/{total_bytes} bytes)")

tracer.close(blocking=True, on_progress=progress_callback)

# Output:
# Upload progress: 25.0% (262144/1048576 bytes)
# Upload progress: 50.0% (524288/1048576 bytes)
# Upload progress: 75.0% (786432/1048576 bytes)
# Upload progress: 100.0% (1048576/1048576 bytes)