Agent integration

backend/agents/database.py

@@ -0,0 +1,113 @@
+from typing import Dict, Any, Optional
+from supabase import create_client
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+SUPABASE_URL = os.getenv('SUPABASE_URL')
+SUPABASE_KEY = os.getenv('SUPABASE_KEY')
+supabase = create_client(SUPABASE_URL, SUPABASE_KEY) if SUPABASE_URL and SUPABASE_KEY else None
+
+
+def _has_client() -> bool:
+	return supabase is not None
+
+
+async def create_result(pr_name: str, pr_link: str, overall_result: Dict[str, Any], run_status: str) -> Optional[int]:
+	"""Insert a new row into results and return its id."""
+	try:
+		if not _has_client():
+			print("[db] Skipping create_result: SUPABASE not configured")
+			return None
+		payload = {
+			"pr_name": pr_name,
+			"pr_link": pr_link,
+			"overall_result": overall_result,
+			"run_status": run_status,
+		}
+		resp = supabase.table('results').insert(payload).execute()
+		row = (resp.data or [{}])[0]
+		print(f"[db] Created result id={row.get('id')} status={run_status}")
+		return row.get('id')
+	except Exception as e:
+		print(f"[db] ❌ create_result error: {str(e)}")
+		return None
+
+
+async def set_suite_result_id(suite_id: int, result_id: int) -> None:
+	"""Link a suite to a result by setting suites.result_id."""
+	try:
+		if not _has_client():
+			print(f"[db] Skipping set_suite_result_id for suite {suite_id}: no client")
+			return
+		supabase.table('suites').update({"result_id": result_id}).eq('id', suite_id).execute()
+		print(f"[db] Linked suite {suite_id} -> result {result_id}")
+	except Exception as e:
+		print(f"[db] ❌ set_suite_result_id error: {str(e)}")
+
+
+async def get_or_create_test(suite_id: int, name: str) -> Optional[int]:
+	"""Find a test row by (suite_id, name) or create it. Returns test id."""
+	try:
+		if not _has_client():
+			print(f"[db] Skipping get_or_create_test for suite {suite_id}, name '{name}': no client")
+			return None
+		# Try find existing
+		resp = supabase.table('tests').select('id').eq('suite_id', suite_id).eq('name', name).limit(1).execute()
+		if resp.data:
+			return resp.data[0]['id']
+		# Create new
+		payload = {
+			"suite_id": suite_id,
+			"name": name,
+			"steps": [],
+			"run_status": "RUNNING",
+		}
+		ins = supabase.table('tests').insert(payload).execute()
+		row = (ins.data or [{}])[0]
+		print(f"[db] Created test id={row.get('id')} for suite {suite_id}, name '{name}'")
+		return row.get('id')
+	except Exception as e:
+		print(f"[db] ❌ get_or_create_test error: {str(e)}")
+		return None
+
+
+async def append_test_step(test_id: int, step: Any) -> None:
+	"""Append a single step to tests.steps immediately (read-modify-write)."""
+	try:
+		if not _has_client():
+			return
+		res = supabase.table('tests').select('steps').eq('id', test_id).limit(1).execute()
+		steps = []
+		if res.data:
+			curr = res.data[0].get('steps')
+			if isinstance(curr, list):
+				steps = curr
+		steps.append(step)
+		supabase.table('tests').update({"steps": steps}).eq('id', test_id).execute()
+	except Exception as e:
+		print(f"[db] ❌ append_test_step error: {str(e)}")
+
+
+async def update_test_fields(test_id: int, fields: Dict[str, Any]) -> None:
+	"""Generic update of tests row."""
+	try:
+		if not _has_client():
+			return
+		supabase.table('tests').update(fields).eq('id', test_id).execute()
+	except Exception as e:
+		print(f"[db] ❌ update_test_fields error: {str(e)}")
+
+
+async def get_result_id_for_suite(suite_id: int) -> Optional[int]:
+	"""Return result_id for a given suite_id from suites table."""
+	try:
+		if not _has_client():
+			return None
+		resp = supabase.table('suites').select('result_id').eq('id', suite_id).limit(1).execute()
+		if resp.data:
+			return resp.data[0].get('result_id')
+		return None
+	except Exception as e:
+		print(f"[db] ❌ get_result_id_for_suite error: {str(e)}")
+		return None

backend/agents/prompts.py

@@ -0,0 +1,63 @@
+from typing import List, Dict
+import os
+
+from dotenv import load_dotenv
+load_dotenv()
+
+def build_agent_instructions(tests: List[Dict], suite: Dict) -> str:
+	"""Build optimized instructions for autonomous agent testing"""
+
+	base_url = os.getenv('DEPLOYMENT_URL', 'https://staging.example.com')
+
+	# Create comprehensive testing instructions
+	instructions = f"""
+You are an autonomous QA testing agent for web applications. Your goal is to thoroughly test the deployment at {base_url}.
+
+SUITE: {suite['name']}
+TOTAL TESTS: {len(tests)}
+
+TESTING APPROACH:
+1. Start by taking a screenshot to see the current state
+2. Navigate to the base URL: {base_url}
+3. Perform comprehensive exploratory testing
+4. Look for bugs, broken functionality, and edge cases
+5. Test user flows and interactions
+6. Pay special attention to recent changes that might have introduced issues
+
+SPECIFIC TEST SCENARIOS:
+"""
+
+	for i, test in enumerate(tests, 1):
+		instructions += f"""
+{i}. {test['name']}
+   Description: {test.get('summary', 'No description provided')}
+   Priority: {'HIGH' if 'critical' in test.get('summary', '').lower() else 'MEDIUM'}
+"""
+
+	instructions += """
+
+TESTING GUIDELINES:
+- Be thorough and methodical in your approach
+- Take screenshots at key moments to document your findings
+- Test both happy paths and edge cases
+- Look for unexpected behaviors, errors, or broken functionality
+- Pay attention to UI/UX issues and usability problems
+- Test form submissions, navigation, and interactive elements
+- Check for responsive design and mobile compatibility if applicable
+- Document any bugs or issues you discover with clear descriptions
+
+SUCCESS CRITERIA:
+- Complete testing of all specified scenarios
+- Identify and document any bugs or issues found
+- Verify that core functionality works as expected
+- Provide clear feedback on the overall quality of the deployment
+
+Remember: You are looking for unexpected bugs and issues that developers might miss. Be creative in your testing approach and explore edge cases.
+
+FINAL VERDICT FORMAT (MANDATORY):
+- After completing each test scenario, output exactly one line with no extra commentary:
+- RESULT: PASSED   (if the scenario executed successfully and no critical issues were found)
+- RESULT: FAILED   (if execution could not complete or a critical/blocking issue was found)
+"""
+
+	return instructions

backend/record_lib.py → backend/agents/record.py

@@ -12,14 +12,6 @@
   - State is stored at /tmp/cua_recorder/state.json
 """
 
-import os
-import json
-import time
-import signal
-import platform
-import subprocess
-from pathlib import Path
-
 
 def start_recording(output_dir=None, fps=None, width=None, height=None, display=None):
     """Start ffmpeg screen recording in background and persist PID.
@@ -115,6 +107,8 @@ def stop_recording(upload_url=None):
     import time as _time
     import signal as _signal
     from pathlib import Path as _Path
+    import urllib.request as _urlreq
+    import uuid as _uuid
 
     state_path = _Path("/tmp/cua_recorder/state.json")
     data = {}
@@ -153,14 +147,13 @@ def stop_recording(upload_url=None):
         # Try upload to server
         upload = {"ok": False}
         try:
-            import urllib.request as _urlreq
-            import uuid as _uuid
-            import os as _os
+            VIDEO_UPLOAD_URL = "https://qai-ashy.vercel.app/upload-video"
 
             # Determine upload URL: param overrides env, skip if none
             _url = upload_url
             if not _url:
-                _url = _os.getenv("VIDEO_UPLOAD_URL")
+                _url = VIDEO_UPLOAD_URL
+            print(f"Upload URL: {_url}")
 
             if path and _os.path.exists(path) and _url:
                 boundary = f"----WebKitFormBoundary{_uuid.uuid4().hex}"
@@ -177,7 +170,7 @@ def stop_recording(upload_url=None):
                 body_prefix = ("".join(parts)).encode("utf-8")
                 body_suffix = (CRLF + f"--{boundary}--{CRLF}").encode("utf-8")
                 body = body_prefix + file_bytes + body_suffix
-
+                print(f"Body: {body}")
                 req = _urlreq.Request(
                     url=_url,
                     data=body,
@@ -196,9 +189,10 @@ def stop_recording(upload_url=None):
                     upload = {"ok": True, "response": upload_json}
         except Exception as _e:
             upload = {"ok": False, "error": repr(_e)}
-
+        print(f"Upload: {upload}")
         return {"ok": True, "path": path, "upload": upload}
     except Exception as e:
+        print(f"Error: {e}")
         return {"ok": False, "error": repr(e)}
 
 

backend/agents/run_suite.py

@@ -5,7 +5,6 @@
 """
 import sys
 import asyncio
-import os
 from pathlib import Path
 
 # Add the agents directory to the Python path