social-app/backend/tests/integration/v1/agent/test_interrupt_resume_flow.py

from __future__ import annotations

from uuid import UUID

import pytest
from fastapi.testclient import TestClient

from app import app
from core.auth.models import CurrentUser
from v1.agent.dependencies import get_agent_service
from v1.agent.schemas import RunAgentInput
from v1.users.dependencies import get_current_user


class FakeAgentServiceWithInterrupt:
    async def stream_run(self, input_data: RunAgentInput):
        yield 'data: {"type": "RUN_STARTED", "runId": "' + input_data.runId + '"}\n\n'
        yield 'data: {"type": "TEXT_MESSAGE_START", "messageId": "m1"}\n\n'
        yield 'data: {"type": "TEXT_MESSAGE_CONTENT", "delta": "Let me navigate"}\n\n'
        yield 'data: {"type": "TOOL_CALL", "toolName": "ui.navigate_to", "args": {"path": "/home"}}\n\n'
        yield (
            'data: {"type": "RUN_FINISHED", "runId": "'
            + input_data.runId
            + '", "outcome": "interrupt", "interrupt": {"id": "int-1", "reason": "frontend_tool", "payload": {"toolName": "ui.navigate_to", "args": {"path": "/home"}}}}\n\n'
        )

    async def stream_resume(self, run_id: str, input_data: RunAgentInput):
        if input_data.resume and input_data.resume.get("interruptId") == "int-1":
            payload = input_data.resume.get("payload", {})
            yield 'data: {"type": "RUN_STARTED", "runId": "' + run_id + '"}\n\n'
            yield (
                'data: {"type": "TOOL_RESULT", "toolName": "ui.navigate_to", "result": '
                + str(payload.get("result", {}))
                + "}\n\n"
            )
            yield 'data: {"type": "TEXT_MESSAGE_START", "messageId": "m2"}\n\n'
            yield 'data: {"type": "TEXT_MESSAGE_CONTENT", "delta": "Navigation completed"}\n\n'
            yield 'data: {"type": "RUN_FINISHED", "runId": "' + run_id + '"}\n\n'
        else:
            yield (
                'data: {"type": "RUN_FINISHED", "runId": "'
                + run_id
                + '", "outcome": "error"}\n\n'
            )


def _get_test_user() -> CurrentUser:
    return CurrentUser(id=UUID("00000000-0000-0000-0000-000000000001"))


@pytest.fixture
def client() -> TestClient:
    app.dependency_overrides[get_current_user] = _get_test_user
    app.dependency_overrides[get_agent_service] = (
        lambda: FakeAgentServiceWithInterrupt()
    )
    yield TestClient(app)
    app.dependency_overrides.clear()


class TestInterruptResumeFlow:
    def test_frontend_tool_interrupt_then_resume_with_result(self, client: TestClient):
        payload = {
            "threadId": "t1",
            "runId": "r1",
            "state": {},
            "messages": [{"role": "user", "content": "Navigate to home"}],
            "tools": [{"name": "ui.navigate_to", "execution_target": "frontend"}],
            "context": [],
            "forwardedProps": {},
        }
        response = client.post("/api/v1/agent/runs", json=payload)
        assert response.status_code == 200

        events = response.text.split("\n\n")
        interrupt_event = [e for e in events if '"outcome": "interrupt"' in e][0]
        assert '"id": "int-1"' in interrupt_event
        assert '"reason": "frontend_tool"' in interrupt_event

        resume_payload = {
            "threadId": "t1",
            "runId": "r1",
            "state": {},
            "messages": [],
            "tools": [],
            "context": [],
            "forwardedProps": {},
            "resume": {
                "interruptId": "int-1",
                "payload": {"result": {"success": True}},
            },
        }
        resume_response = client.post(
            "/api/v1/agent/runs/r1/resume", json=resume_payload
        )
        assert resume_response.status_code == 200

        resume_events = resume_response.text.split("\n\n")
        tool_result_event = [e for e in resume_events if '"type": "TOOL_RESULT"' in e][
            0
        ]
        assert '"toolName": "ui.navigate_to"' in tool_result_event
        assert (
            "'success': True" in tool_result_event
            or '"success": true' in tool_result_event.lower()
        )

    def test_backend_tool_approval_rejected(self, client: TestClient):
        payload = {
            "threadId": "t2",
            "runId": "r2",
            "state": {},
            "messages": [{"role": "user", "content": "Transfer funds"}],
            "tools": [
                {
                    "name": "srv.transfer_funds",
                    "execution_target": "backend",
                    "requires_approval": True,
                }
            ],
            "context": [],
            "forwardedProps": {},
        }
        response = client.post("/api/v1/agent/runs", json=payload)
        assert response.status_code == 200

        resume_payload = {
            "threadId": "t2",
            "runId": "r2",
            "state": {},
            "messages": [],
            "tools": [],
            "context": [],
            "forwardedProps": {},
            "resume": {
                "interruptId": "int-1",
                "payload": {"decision": "rejected", "reason": "User denied"},
            },
        }
        resume_response = client.post(
            "/api/v1/agent/runs/r2/resume", json=resume_payload
        )
        assert resume_response.status_code == 200