import pytest import httpx import os import uuid import time from conftest import BASE_URL def _headers(): uid = os.getenv("SYNC_TEST_USER_ID", "") return {"X-User-ID": uid} def test_coworker_sc1_mirror_check(): """ SC-1 (Mirror Check): 1. Deploy an agent with co_worker_quality_gate=True. 2. Wait for the agent to initialize (Status: evaluating). 3. Use the /nodes/{id}/fs/ls API to verify the .cortex folder existence. """ node_id = f"test-coworker-sc1-{uuid.uuid4().hex[:8]}" admin_id = os.getenv("SYNC_TEST_USER_ID", "") instance_id = None with httpx.Client(timeout=30.0) as client: try: # 1. Register a test node node_payload = { "node_id": node_id, "display_name": "Co-Worker SC-1 Node", "is_active": True, "skill_config": {"shell": {"enabled": True}, "sync": {"enabled": True}} } r_node = client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload) assert r_node.status_code == 200, f"Node registration failed: {r_node.text}" # 2. Deploy Agent with co_worker_quality_gate=True deploy_payload = { "name": "SC-1 Mirror Agent", "description": "Tests .cortex mirror initialization", "system_prompt": "You are a test agent. Create a simple hello world python script.", "max_loop_iterations": 1, "mesh_node_id": node_id, "provider_name": "gemini", "model_name": "gemini-1.5-flash", # Explicitly use flash "trigger_type": "interval", "interval_seconds": 60, # Long interval so it doesn't run twice during test "co_worker_quality_gate": True, "default_prompt": "Create app.py that prints hello.", } r_deploy = client.post(f"{BASE_URL}/agents/deploy", json=deploy_payload, headers=_headers()) assert r_deploy.status_code == 200, f"Deploy failed: {r_deploy.text}" instance_id = r_deploy.json()["instance_id"] # 3. Wait for agent to initialize (Status: evaluating) print(f"\n[test] Waiting for agent {instance_id} to reach 'evaluating' status...") found_evaluating = False for _ in range(30): # 60s timeout r_agent = client.get(f"{BASE_URL}/agents/{instance_id}", headers=_headers()) if r_agent.status_code == 200: agent = r_agent.json() if agent.get("evaluation_status") == "evaluating": found_evaluating = True break time.sleep(2) assert found_evaluating, f"Agent did not reach 'evaluating' status." # 4. Use the /nodes/{id}/fs/ls API to verify the .cortex folder existence r_ls = client.get(f"{BASE_URL}/nodes/{node_id}/fs/ls", params={"path": ".cortex"}, headers=_headers()) assert r_ls.status_code == 200, f"Failed to ls .cortex: {r_ls.text}" files = r_ls.json() filenames = [f["name"] for f in files] # Verify rubric.md and history.log are present as per test plan assert any("rubric.md" in f for f in filenames), f"rubric.md not found in {filenames}" assert any("history.log" in f for f in filenames), f"history.log not found in {filenames}" finally: if instance_id: client.delete(f"{BASE_URL}/agents/{instance_id}", headers=_headers()) client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id}) def test_coworker_sc3_limit_check(): """ SC-3 (Limit Check): 1. Deploy an agent with max_rework_attempts=1 and rework_threshold=100. 2. Trigger a run. 3. Poll the /agents endpoint until evaluation_status == 'failed_limit'. 4. Verify the latest_quality_score is present in the response. """ node_id = f"test-coworker-sc3-{uuid.uuid4().hex[:8]}" admin_id = os.getenv("SYNC_TEST_USER_ID", "") instance_id = None with httpx.Client(timeout=30.0) as client: try: # 1. Register a test node node_payload = { "node_id": node_id, "display_name": "Co-Worker SC-3 Node", "is_active": True, "skill_config": {"shell": {"enabled": True}, "sync": {"enabled": True}} } r_node = client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload) assert r_node.status_code == 200, f"Node registration failed: {r_node.text}" # 2. Deploy Agent with max_rework_attempts=1 and rework_threshold=100 deploy_payload = { "name": "SC-3 Limit Agent", "system_prompt": "You are a test agent. Create a simple hello world python script.", "max_loop_iterations": 1, "mesh_node_id": node_id, "provider_name": "gemini", "model_name": "gemini-1.5-flash", "trigger_type": "webhook", # Use webhook to trigger manually "co_worker_quality_gate": True, "max_rework_attempts": 1, "rework_threshold": 100, # Impossible to pass "default_prompt": "Create app.py that prints hello.", } r_deploy = client.post(f"{BASE_URL}/agents/deploy", json=deploy_payload, headers=_headers()) assert r_deploy.status_code == 200, f"Deploy failed: {r_deploy.text}" instance_id = r_deploy.json()["instance_id"] # 3. Get the webhook secret and trigger it r_trig = client.get(f"{BASE_URL}/agents/{instance_id}/triggers", headers=_headers()) webhook_trigger = next(t for t in r_trig.json() if t["trigger_type"] == "webhook") secret = webhook_trigger["webhook_secret"] r_hook = client.post( f"{BASE_URL}/agents/{instance_id}/webhook", params={"token": secret}, json={"prompt": "Go!"} ) assert r_hook.status_code == 202 # 4. Poll until evaluation_status == 'failed_limit' print(f"\n[test] Waiting for agent {instance_id} to reach 'failed_limit' status...") failed_limit = False latest_score = None for _ in range(60): # 120s timeout r_agents = client.get(f"{BASE_URL}/agents", headers=_headers()) if r_agents.status_code == 200: agents = r_agents.json() agent = next((a for a in agents if a["id"] == instance_id), None) if agent: status = agent.get("evaluation_status") latest_score = agent.get("latest_quality_score") if status == "failed_limit": failed_limit = True break time.sleep(2) assert failed_limit, f"Agent did not reach 'failed_limit' status." assert latest_score is not None, "latest_quality_score should be present in the response" finally: if instance_id: client.delete(f"{BASE_URL}/agents/{instance_id}", headers=_headers()) client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id})