agent-platform/tests/test_team_service.py

from pathlib import Path
from datetime import datetime

from tests.conftest import (
    build_fastapi_test_client,
    build_postgres_database_url,
    build_postgres_engine,
    prepare_known_service_import,
)


def test_team_service_post_contract_supports_team_configs_and_runs(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("team-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from core_db import create_session_factory

    database_url = build_postgres_database_url(tmp_path, "teams-api")
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    monkeypatch.setenv("AGENT_PLATFORM_REDIS_URL", "")
    monkeypatch.setenv("AGENT_PLATFORM_AUTO_WORKER_ENABLED", "false")
    engine = build_postgres_engine(database_url)
    Base.metadata.create_all(engine)

    app = create_app()
    app.state.session_factory = create_session_factory(engine)
    client = build_fastapi_test_client(app)

    team_response = client.post(
        "/teams/create",
        json={
            "name": "Support Team",
            "description": "Handles support escalations",
            "teamType": "collaborative",
            "ownerUserId": "demo-user",
        },
    )
    assert team_response.status_code == 200
    team_payload = team_response.json()["data"]
    assert team_payload["name"] == "Support Team"
    assert team_payload["teamType"] == "collaborative"
    assert "code" not in team_payload

    config_response = client.post(
        "/teams/configs/create",
        json={
            "teamId": team_payload["id"],
            "coordinationMode": "supervisor",
            "objective": "Resolve the customer request",
            "memberRefs": [
                {
                    "role": "worker",
                    "agentId": "agent_support",
                    "responsibility": "Draft the answer",
                }
            ],
            "policy": {
                "max_rounds": 3,
                "handoff": "supervisor",
            },
        },
    )
    assert config_response.status_code == 200
    config_payload = config_response.json()["data"]
    assert config_payload["teamId"] == team_payload["id"]
    assert config_payload["memberRefs"][0]["role"] == "executor"
    assert config_payload["memberRefs"][0]["member_key"] == "member_1"

    list_response = client.post(
        "/teams/list",
        json={"page": 1, "pageSize": 20, "keyword": "support"},
    )
    assert list_response.status_code == 200
    assert list_response.json()["data"]["total"] == 1

    configs_response = client.post(
        "/teams/configs/list",
        json={"page": 1, "pageSize": 20, "teamId": team_payload["id"]},
    )
    assert configs_response.status_code == 200
    assert configs_response.json()["data"]["total"] == 1

    run_response = client.post(
        "/teams/runs/create",
        json={
            "teamId": team_payload["id"],
            "teamConfigId": config_payload["id"],
            "inputText": "Help the customer reset MFA",
        },
    )
    assert run_response.status_code == 200
    run_payload = run_response.json()["data"]
    assert run_payload["teamId"] == team_payload["id"]
    assert run_payload["teamConfigId"] == config_payload["id"]
    assert run_payload["status"] == "queued"

    status_response = client.post(
        "/teams/runs/status",
        json={
            "teamRunId": run_payload["id"],
            "status": "completed",
            "workerKey": "test-worker",
            "outputText": "MFA reset steps prepared.",
        },
    )
    assert status_response.status_code == 200
    assert status_response.json()["data"]["status"] == "completed"

    runs_response = client.post(
        "/teams/runs/list",
        json={"page": 1, "pageSize": 20, "teamId": team_payload["id"]},
    )
    assert runs_response.status_code == 200
    assert runs_response.json()["data"]["total"] == 1

    update_response = client.post(
        "/teams/update",
        json={
            "teamId": team_payload["id"],
            "name": "Support Team Updated",
            "status": "active",
        },
    )
    assert update_response.status_code == 200
    assert update_response.json()["data"]["name"] == "Support Team Updated"
    assert update_response.json()["data"]["status"] == "active"

    delete_run_response = client.post(
        "/teams/runs/delete",
        json={"teamRunId": run_payload["id"]},
    )
    assert delete_run_response.status_code == 200
    assert delete_run_response.json()["data"]["deleted"] is True

    delete_config_response = client.post(
        "/teams/configs/delete",
        json={"configId": config_payload["id"]},
    )
    assert delete_config_response.status_code == 200
    assert delete_config_response.json()["data"]["deleted"] is True

    delete_team_response = client.post(
        "/teams/delete",
        json={"teamId": team_payload["id"]},
    )
    assert delete_team_response.status_code == 200
    assert delete_team_response.json()["data"]["deleted"] is True


def test_team_service_compacts_member_context_between_agent_calls() -> None:
    prepare_known_service_import("team-service")

    from app.application.services import TeamApplicationService, TeamMemberRunResult
    from core_domain import AgentRunContract, TeamMemberContract

    service = TeamApplicationService(
        team_repository=None,
        team_config_repository=None,
        team_run_repository=None)
    result = TeamMemberRunResult(
        member=TeamMemberContract(
            member_key="member_1",
            agent_id="agent_1",
            role="supervisor",
            name="Planner"),
        run=AgentRunContract(
            id="run_1",
            agent_id="agent_1",
            agent_config_id="config_1",
            output_text="x" * 2000,
            output_json={
                "model": "demo-model",
                "finish_reason": "stop",
                "messages": [{"role": "user", "content": "large prompt"}],
                "raw_response_json": {"thinking": "hidden debug payload"},
            },
            status="completed",
            created_time=datetime.utcnow()))

    prior_output = service._compact_prior_output(result)
    member_json = service._member_result_to_json(result)

    assert prior_output["output_text"].endswith("[truncated]")
    assert prior_output["output_json"] == {
        "model": "demo-model",
        "finish_reason": "stop",
        "debug_payload_omitted": True,
    }
    assert "messages" not in member_json["output_json"]
    assert "raw_response_json" not in member_json["output_json"]