import asyncio import hashlib import uuid from datetime import UTC, datetime, timedelta from app.computer_client import ( ComputerWorkerActionResult, ComputerWorkerSessionState, get_computer_client, ) from app.errors import AppError from app.health import get_computer_health_probe from app.main import app from app.models import ( Agent, ApprovalRequest, AuditEvent, Computer, ComputerAction, ComputerArtifact, ComputerControlLease, ComputerProfile, ComputerSession, IdempotencyRecord, User, Workspace, WorkspaceMembership, ) from sqlalchemy import select def owner_payload(): return { "email": "owner@example.com", "password": "correct horse battery staple", "display_name": "Owner", "workspace_name": "Personal", "timezone": "Asia/Dubai", } def csrf_headers(client): return {"x-csrf-token": client.cookies.get("hayva_csrf")} def action_headers(client): return { **csrf_headers(client), "x-idempotency-key": f"test-action-{uuid.uuid4()}", } class HealthyComputerProbe: async def status(self): return "configured" class FakeComputerClient: def __init__(self, workspace_id: str | None = None): self.calls: list[tuple[str, int]] = [] self.request_ids: list[str] = [] self.workspace_id = workspace_id self.artifact_content = b"verified fake jpeg bytes" def state(self, status: str, *, changed: bool = False): return ComputerWorkerSessionState( worker_id="worker-test-1", status=status, current_url=( "https://example.com/after-human" if changed else "https://example.com/start" ), active_tab_id="tab-1", tabs=[{"id": "tab-1", "url": "https://example.com/start"}], page_title="Example", observed_at=datetime.now(UTC).isoformat(), ) async def start_session(self, **kwargs): self.request_ids.append(kwargs["request_id"]) self.calls.append(("start", kwargs["fencing_token"])) return self.state("ai_controlled") async def command(self, **kwargs): self.request_ids.append(kwargs["request_id"]) command = kwargs["command"] self.calls.append((command, kwargs["fencing_token"])) status = {"pause": "paused", "resume": "ai_controlled", "stop": "stopped"}[command] return self.state(status, changed=command == "resume") async def snapshot(self, **kwargs): self.request_ids.append(kwargs["request_id"]) self.calls.append(("snapshot", kwargs["fencing_token"])) return self.state("human_controlled", changed=True) async def action(self, **kwargs): self.calls.append((f"action:{kwargs['tool']}", kwargs["fencing_token"])) if kwargs["tool"] == "observe": result = { "url": "https://example.com/start", "title": "Example", "interactive_elements": [{"tag": "button", "text": "Save"}], } elif kwargs["tool"] == "extract": result = {"value": "authorized private result"} elif kwargs["tool"] == "screenshot" and self.workspace_id: artifact_id = uuid.uuid4() result = {"artifact": { "id": str(artifact_id), "storage_key": ( f"{self.workspace_id}/{kwargs['session_id']}/{artifact_id}.jpg" ), "sha256": hashlib.sha256(self.artifact_content).hexdigest(), "mime_type": "image/jpeg", "size_bytes": len(self.artifact_content), "metadata": {"width": 1440, "height": 900}, }} else: result = {"accepted": True} return ComputerWorkerActionResult( session=self.state("ai_controlled"), tool=kwargs["tool"], result=result, verification={"status": "observed", "url": "https://example.com/start"}, ) async def human_pointer(self, **kwargs): self.calls.append(("human:pointer", kwargs["fencing_token"])) return self.state("human_controlled", changed=True) async def human_keyboard(self, **kwargs): self.calls.append(("human:keyboard", kwargs["fencing_token"])) return self.state("human_controlled", changed=True) async def frame(self, **kwargs): self.calls.append(("frame", kwargs["fencing_token"])) return { "session_id": str(kwargs["session_id"]), "fencing_token": kwargs["fencing_token"], "image_base64": "dGVzdA==", "mime_type": "image/jpeg", "width": 1440, "height": 900, } async def artifact(self, **kwargs): return self.artifact_content async def delete_artifact(self, **kwargs): self.calls.append((f"delete:{kwargs['artifact_id']}", 0)) class UnavailableComputerClient(FakeComputerClient): async def start_session(self, **kwargs): raise AppError("COMPUTER_SERVICE_UNAVAILABLE", "The computer service is unavailable.", 503) async def command(self, **kwargs): raise AppError("COMPUTER_SERVICE_UNAVAILABLE", "The computer service is unavailable.", 503) class UncertainActionClient(FakeComputerClient): async def action(self, **kwargs): raise AppError( "COMPUTER_SERVICE_UNAVAILABLE", "The computer service is unavailable.", 503 ) def install_computer_overrides(fake): app.dependency_overrides[get_computer_client] = lambda: fake app.dependency_overrides[get_computer_health_probe] = lambda: HealthyComputerProbe() def prepare_browser_computer(client, fake): install_computer_overrides(fake) async def activate(): async with client.session_factory() as session: agent = await session.scalar(select(Agent)) computer = await session.get(Computer, agent.computer_id) profile = await session.scalar(select(ComputerProfile).where( ComputerProfile.agent_id == agent.id )) computer.status = "running" computer.provider_resource_id = f"test-vm-{computer.id}" computer.compute_provider = "test" agent.status = "active" profile.status = "active" await session.commit() asyncio.run(activate()) def start_payload(profile_id): return { "profile_id": profile_id, "task_summary": "Open the workspace portal and inspect its current state.", } def test_computer_session_takeover_return_pause_resume_and_stop(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = FakeComputerClient() prepare_browser_computer(auth_client, fake) status = auth_client.get("/api/v1/computer/status") assert status.status_code == 200 assert status.json()["service_state"] == "configured" profiles = auth_client.get("/api/v1/computer/profiles").json()["profiles"] assert len(profiles) == 1 assert "storage_key" not in profiles[0] missing_csrf = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profiles[0]["id"]) ) assert missing_csrf.status_code == 403 created = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profiles[0]["id"]), headers=csrf_headers(auth_client), ) assert created.status_code == 201 computer = created.json()["session"] assert computer["status"] == "ai_controlled" assert computer["control_owner"] == "ai" assert fake.calls == [("start", 1)] takeover = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/takeover", headers=csrf_headers(auth_client), ) assert takeover.status_code == 200 assert takeover.json()["session"]["status"] == "human_controlled" assert fake.calls[-1] == ("pause", 1) async def set_human_controller(actor_user_id): async with auth_client.session_factory() as session: lease = await session.scalar(select(ComputerControlLease).where( ComputerControlLease.session_id == uuid.UUID(computer["id"]), ComputerControlLease.released_at.is_(None), )) lease.actor_user_id = actor_user_id await session.commit() asyncio.run(set_human_controller(None)) denied_pause = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/pause", headers=csrf_headers(auth_client), ) assert denied_pause.status_code == 403 assert denied_pause.json()["error"]["code"] == "COMPUTER_HUMAN_CONTROL_OWNED" owner_id = auth_client.get("/api/v1/auth/me").json()["user"]["id"] asyncio.run(set_human_controller(uuid.UUID(owner_id))) returned = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/return-control", headers=csrf_headers(auth_client), ) assert returned.status_code == 200 assert returned.json()["reobserved"] is True assert returned.json()["session"]["current_url"].endswith("after-human") assert fake.calls[-2:] == [("snapshot", 2), ("resume", 3)] assert fake.request_ids[-2] != fake.request_ids[-1] assert fake.request_ids[-2].endswith(".snapshot") assert fake.request_ids[-1].endswith(".resume") paused = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/pause", headers=csrf_headers(auth_client), ) assert paused.status_code == 200 assert paused.json()["session"]["status"] == "paused" resumed = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/resume", headers=csrf_headers(auth_client), ) assert resumed.status_code == 200 assert resumed.json()["session"]["status"] == "ai_controlled" stopped = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/stop", headers=csrf_headers(auth_client), ) assert stopped.status_code == 200 assert stopped.json()["session"]["status"] == "stopped" assert stopped.json()["session"]["control_owner"] == "none" detail = auth_client.get(f"/api/v1/computer/sessions/{computer['id']}").json() assert detail["control_lease"] is None checkpoint_types = {item["checkpoint_type"] for item in detail["checkpoints"]} assert {"session_started", "takeover", "return_control", "pause", "recovery", "completion"} <= checkpoint_types async def lease_history(): async with auth_client.session_factory() as session: return list((await session.scalars(select(ComputerControlLease).where( ComputerControlLease.session_id == uuid.UUID(computer["id"]) ).order_by(ComputerControlLease.fencing_token))).all()) leases = asyncio.run(lease_history()) assert [lease.fencing_token for lease in leases] == [1, 2, 3, 4] assert all(lease.released_at for lease in leases) def test_emergency_stop_blocks_new_sessions_and_never_auto_resumes(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = FakeComputerClient() prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] created = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] emergency = auth_client.post( "/api/v1/computer/emergency-stop", json={"reason": "Owner requested an immediate safety stop."}, headers=csrf_headers(auth_client), ) assert emergency.status_code == 200 assert emergency.json()["verified_stopped_sessions"] == 1 blocked = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ) assert blocked.status_code == 409 assert blocked.json()["error"]["code"] == "COMPUTER_EMERGENCY_STOPPED" assert auth_client.get( f"/api/v1/computer/sessions/{created['id']}" ).json()["session"]["status"] == "stopped" resumed = auth_client.post( "/api/v1/computer/emergency-resume", headers=csrf_headers(auth_client) ) assert resumed.status_code == 200 assert resumed.json()["sessions_auto_resumed"] is False assert auth_client.get( f"/api/v1/computer/sessions/{created['id']}" ).json()["session"]["status"] == "stopped" def test_unavailable_worker_is_persisted_as_failure_not_success(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) prepare_browser_computer(auth_client, UnavailableComputerClient()) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] response = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ) assert response.status_code == 503 assert response.json()["error"]["code"] == "COMPUTER_SERVICE_UNAVAILABLE" records = auth_client.get("/api/v1/computer/sessions").json()["sessions"] assert records[0]["status"] == "failed" assert records[0]["failure_code"] == "COMPUTER_SERVICE_UNAVAILABLE" assert records[0]["control_owner"] == "none" def test_computer_session_detail_cannot_cross_workspace(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) other_session_id = uuid.uuid4() async def create_other_workspace_session(): async with auth_client.session_factory() as session: user = await session.scalar(select(User)) workspace = Workspace(id=uuid.uuid4(), name="Other", timezone="UTC") session.add(workspace) await session.flush() session.add(WorkspaceMembership( id=uuid.uuid4(), workspace_id=workspace.id, user_id=user.id, status="active" )) await session.flush() agent = Agent( id=uuid.uuid4(), workspace_id=workspace.id, created_by_user_id=user.id, name="Private Other Agent", description="", purpose="", status="disabled", autonomy_level=1, computer_required=False, budget_currency="USD", version=1, ) profile = ComputerProfile( id=uuid.uuid4(), workspace_id=workspace.id, created_by_user_id=user.id, profile_key="personal", name="Other Private Computer", storage_key=uuid.uuid4().hex, status="revoked", retention_days=30, version=1, ) session.add_all([agent, profile]) await session.flush() session.add(ComputerSession( id=other_session_id, workspace_id=workspace.id, requested_by_user_id=user.id, agent_id=agent.id, profile_id=profile.id, task_summary="Private task", status="stopped", control_owner="none", lease_version=0, version=1, )) await session.commit() asyncio.run(create_other_workspace_session()) response = auth_client.get(f"/api/v1/computer/sessions/{other_session_id}") assert response.status_code == 404 assert response.json()["error"]["code"] == "COMPUTER_SESSION_NOT_FOUND" def test_browser_actions_are_grant_checked_approval_bound_redacted_and_verified(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = FakeComputerClient() prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] computer = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] observe_headers = action_headers(auth_client) observed = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "observe", "arguments": {}}, headers=observe_headers, ) assert observed.status_code == 200 assert observed.json()["action"]["status"] == "succeeded" assert observed.json()["action"]["verification"]["status"] == "observed" replayed = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "observe", "arguments": {}}, headers=observe_headers, ) assert replayed.status_code == 200 assert replayed.json()["action"]["id"] == observed.json()["action"]["id"] assert [name for name, _ in fake.calls].count("action:observe") == 1 secret = "never persist this secret" proposed = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "type", "arguments": {"selector": "#password", "text": secret}}, headers=action_headers(auth_client), ) assert proposed.status_code == 202 action = proposed.json()["action"] assert action["status"] == "awaiting_approval" assert secret not in str(action) pending = auth_client.get("/api/v1/computer/approvals").json()["approvals"] assert pending[0]["action_id"] == action["id"] assert secret not in str(pending[0]) mismatch = auth_client.post( f"/api/v1/computer/actions/{action['id']}/decision", json={"approve": True, "arguments": {"selector": "#password", "text": "changed"}}, headers=csrf_headers(auth_client), ) assert mismatch.status_code == 409 assert mismatch.json()["error"]["code"] == "COMPUTER_APPROVAL_PAYLOAD_MISMATCH" approved = auth_client.post( f"/api/v1/computer/actions/{action['id']}/decision", json={"approve": True, "arguments": {"selector": "#password", "text": secret}}, headers=csrf_headers(auth_client), ) assert approved.status_code == 200 assert approved.json()["action"]["status"] == "succeeded" assert secret not in str(approved.json()["action"]) assert fake.calls[-1][0] == "action:type" def test_live_frame_and_human_inputs_require_the_human_lease_owner(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = FakeComputerClient() prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] computer = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/takeover", headers=csrf_headers(auth_client), ) frame = auth_client.get(f"/api/v1/computer/sessions/{computer['id']}/frame") assert frame.status_code == 200 assert frame.headers["cache-control"] == "no-store" assert frame.json()["frame"]["mime_type"] == "image/jpeg" pointer = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/human/pointer", json={"action": "click", "x": 100, "y": 200}, headers=csrf_headers(auth_client), ) assert pointer.status_code == 200 keyboard = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/human/keyboard", json={"text": "human private text"}, headers=csrf_headers(auth_client), ) assert keyboard.status_code == 200 detail = auth_client.get(f"/api/v1/computer/sessions/{computer['id']}").json() human_actions = [item for item in detail["actions"] if item["actor_type"] == "user"] assert len(human_actions) == 2 assert "human private text" not in str(human_actions) def test_screenshot_artifact_is_workspace_scoped_and_integrity_verified(auth_client): setup = auth_client.post("/api/v1/setup/owner", json=owner_payload()).json() fake = FakeComputerClient(setup["workspace_id"]) prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] computer = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] captured = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "screenshot", "arguments": {}}, headers=action_headers(auth_client), ) assert captured.status_code == 200 artifact_id = captured.json()["action"]["result"]["artifact_id"] downloaded = auth_client.get(f"/api/v1/computer/artifacts/{artifact_id}") assert downloaded.status_code == 200 assert downloaded.content == fake.artifact_content assert downloaded.headers["content-type"] == "image/jpeg" assert downloaded.headers["cache-control"] == "no-store" def test_failed_worker_session_can_reopen_its_persistent_profile(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) prepare_browser_computer(auth_client, UnavailableComputerClient()) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] failed = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ) assert failed.status_code == 503 session_id = auth_client.get("/api/v1/computer/sessions").json()["sessions"][0]["id"] recovered_worker = FakeComputerClient() install_computer_overrides(recovered_worker) recovered = auth_client.post( f"/api/v1/computer/sessions/{session_id}/recover", headers=csrf_headers(auth_client), ) assert recovered.status_code == 200 assert recovered.json()["recovered"] is True assert recovered.json()["session"]["status"] == "ai_controlled" assert recovered_worker.calls == [("start", 2)] detail = auth_client.get(f"/api/v1/computer/sessions/{session_id}").json() assert "recovery" in {item["checkpoint_type"] for item in detail["checkpoints"]} def test_uncertain_browser_write_pauses_until_human_reconciliation(auth_client): auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = UncertainActionClient() prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] computer = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] proposal = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "click", "arguments": {"selector": "#submit"}}, headers=action_headers(auth_client), ) assert proposal.status_code == 202 action_id = proposal.json()["action"]["id"] decision = auth_client.post( f"/api/v1/computer/actions/{action_id}/decision", json={"approve": True, "arguments": {"selector": "#submit"}}, headers=csrf_headers(auth_client), ) assert decision.status_code == 409 assert decision.json()["error"]["code"] == "COMPUTER_ACTION_OUTCOME_UNKNOWN" detail = auth_client.get( f"/api/v1/computer/sessions/{computer['id']}" ).json() assert detail["session"]["status"] == "paused" unknown = next(item for item in detail["actions"] if item["id"] == action_id) assert unknown["status"] == "unknown" assert detail["control_lease"] is None reconciled = auth_client.post( f"/api/v1/computer/actions/{action_id}/reconcile", json={ "outcome": "succeeded", "evidence_summary": "The submitted record is visible once in external history.", }, headers=csrf_headers(auth_client), ) assert reconciled.status_code == 200 assert reconciled.json()["action"]["status"] == "succeeded" repeated = auth_client.post( f"/api/v1/computer/actions/{action_id}/reconcile", json={"outcome": "failed", "evidence_summary": "A changed decision is unsafe."}, headers=csrf_headers(auth_client), ) assert repeated.status_code == 409 def test_maintenance_expires_authority_reconciles_actions_and_deletes_artifacts(auth_client): from app.maintenance import run_maintenance_once auth_client.post("/api/v1/setup/owner", json=owner_payload()) fake = FakeComputerClient() prepare_browser_computer(auth_client, fake) profile = auth_client.get("/api/v1/computer/profiles").json()["profiles"][0] computer = auth_client.post( "/api/v1/computer/sessions", json=start_payload(profile["id"]), headers=csrf_headers(auth_client), ).json()["session"] proposal = auth_client.post( f"/api/v1/computer/sessions/{computer['id']}/actions", json={"tool": "click", "arguments": {"selector": "#save"}}, headers=action_headers(auth_client), ).json()["action"] now = datetime.now(UTC) ids: dict[str, uuid.UUID] = {} async def exercise_maintenance(): async with auth_client.session_factory() as session: record = await session.scalar(select(ComputerSession).where( ComputerSession.id == uuid.UUID(computer["id"]) )) lease = await session.scalar(select(ComputerControlLease).where( ComputerControlLease.session_id == record.id, ComputerControlLease.released_at.is_(None), )) approval = await session.scalar(select(ApprovalRequest).where( ApprovalRequest.action_id == uuid.UUID(proposal["id"]) )) lease.expires_at = now - timedelta(seconds=1) approval.expires_at = now - timedelta(seconds=1) write_action = ComputerAction( workspace_id=record.workspace_id, session_id=record.id, actor_type="user", actor_id=record.requested_by_user_id, tool_name="click", parameters_redacted={}, risk="high", access_type="write", status="running", ) read_action = ComputerAction( workspace_id=record.workspace_id, session_id=record.id, actor_type="user", actor_id=record.requested_by_user_id, tool_name="observe", parameters_redacted={}, risk="low", access_type="read", status="running", ) artifact = ComputerArtifact( workspace_id=record.workspace_id, session_id=record.id, kind="screenshot", status="available", storage_key=f"{record.workspace_id}/{record.id}/{uuid.uuid4()}.jpg", sha256="a" * 64, mime_type="image/jpeg", size_bytes=12, metadata_redacted={}, expires_at=now - timedelta(seconds=1), ) completed = IdempotencyRecord( workspace_id=record.workspace_id, key=f"complete-{uuid.uuid4()}", request_hash="b" * 64, status="completed", response_status=200, response_body={"success": True}, expires_at=now - timedelta(seconds=1), ) in_progress = IdempotencyRecord( workspace_id=record.workspace_id, key=f"running-{uuid.uuid4()}", request_hash="c" * 64, status="in_progress", expires_at=now - timedelta(seconds=1), ) session.add_all([write_action, read_action, artifact, completed, in_progress]) await session.flush() ids.update({ "write": write_action.id, "read": read_action.id, "artifact": artifact.id, "in_progress": in_progress.id, }) await session.commit() async with auth_client.session_factory() as session: counts = await run_maintenance_once(session, now=now, client=fake) assert counts == { "leases_released": 1, "approvals_expired": 1, "artifacts_deleted": 1, "artifacts_pending_deletion": 0, "idempotency_deleted": 1, "idempotency_unknown": 1, } async with auth_client.session_factory() as session: refreshed = await session.scalar(select(ComputerSession).where( ComputerSession.id == uuid.UUID(computer["id"]) )) write_action = await session.get(ComputerAction, ids["write"]) read_action = await session.get(ComputerAction, ids["read"]) artifact = await session.get(ComputerArtifact, ids["artifact"]) in_progress = await session.get(IdempotencyRecord, ids["in_progress"]) approval = await session.scalar(select(ApprovalRequest).where( ApprovalRequest.action_id == uuid.UUID(proposal["id"]) )) events = set((await session.scalars(select(AuditEvent.event_type))).all()) assert refreshed.status == "paused" assert refreshed.control_owner == "none" assert write_action.status == "unknown" assert read_action.status == "failed" assert artifact.status == "deleted" assert in_progress.status == "unknown" assert approval.status == "expired" assert {"computer.lease_expired", "computer.approval_expired"} <= events asyncio.run(exercise_maintenance())