smart-org/apps/api/tests/test_direct_tasks_engine.py

"""Direct tasks enter the engine, and the legacy "run" button refuses them (decision D-D).

What these tests defend:

* a task made over HTTP ``POST /tasks`` gets a pipeline state at ``pending_manager_review``
  in the organization's project, so the scheduler's manager takes it from there
* ``POST /tasks/{id}/runs`` on such a task is refused with 409: the engine owns it, and
  nothing runs outside the engine
* a task made through the MCP gateway (``tasks_create``) gets the same state
* a pre-v2 task with no state still runs through the legacy path
"""

from __future__ import annotations

from httpx import AsyncClient

from app.application.services.mcp_service import McpService, TokenRateLimiter
from app.domain.base import new_id
from app.domain.identity.models import PlatformRole, Role
from app.domain.workforce.pipeline import Stage
from app.infrastructure.db.models import MembershipRow, OrganizationRow, UserRow
from app.infrastructure.repositories.engine import PipelineStateRepository
from app.infrastructure.repositories.workforce import TaskRepository
from app.interfaces.mcp.gateway import McpGateway
from tests.conftest import auth_header, signup
from tests.legacy_tasks import legacy_task

#: Part of the Persian refusal shown when the legacy run button is pressed on an engine task.
ENGINE_OWNS = "موتور خودکار"


def _payload(name: str, slug: str) -> dict:
    return {
        "org_name": name,
        "slug": slug,
        "app_name": f"app-{slug}",
        "app_type": "web_app",
        "language": "node",
        "framework": "next",
        "git_source": "platform",
    }


async def _wizard_org(client: AsyncClient, token: str, name: str, slug: str) -> dict:
    """The wizard's answer: the organization, its project and its team."""
    response = await client.post(
        "/api/setup/complete", json=_payload(name, slug), headers=auth_header(token)
    )
    assert response.status_code == 201, response.text
    return response.json()


async def _state(session_factory, task_id: str):
    async with session_factory() as session:
        return await PipelineStateRepository(session).get(task_id)


async def _org_with_owner(session_factory, slug: str) -> tuple[str, str]:
    """An organization and its owner, made directly (the MCP gateway needs no wizard)."""
    async with session_factory() as session:
        user = UserRow(
            id=new_id(),
            email=f"{slug}@example.com",
            password_hash="x",
            name=slug,
            platform_role=PlatformRole.MANAGER.value,
        )
        session.add(user)
        await session.flush()
        org = OrganizationRow(id=new_id(), name=slug, slug=slug, owner_user_id=user.id)
        session.add(org)
        await session.flush()
        session.add(MembershipRow(organization_id=org.id, user_id=user.id, role=Role.OWNER.value))
        await session.commit()
        return org.id, user.id


async def _mcp_token(session_factory, organization_id: str, user_id: str) -> str:
    async with session_factory() as session:
        issued = await McpService(session).issue(
            organization_id=organization_id,
            name="agent",
            scopes=["tasks:write", "tasks:read"],
            created_by_user_id=user_id,
        )
        await session.commit()
        return issued.token


async def test_a_task_made_over_http_enters_the_engine(client, session_factory, seeded):
    manager = await signup(client, "direct-http@example.com")
    body = await _wizard_org(client, manager["access_token"], "سازمان الف", "direct-http")
    org_id = body["organization"]["id"]
    project_id = body["project"]["id"]

    created = await client.post(
        f"/api/organizations/{org_id}/tasks",
        json={"title": "صفحهٔ ورود"},
        headers=auth_header(manager["access_token"]),
    )
    assert created.status_code == 201, created.text
    task_id = created.json()["id"]

    state = await _state(session_factory, task_id)
    assert state is not None, "a direct task must get a pipeline state"
    assert state.organization_id == org_id
    assert state.project_id == project_id
    assert state.stage is Stage.PENDING_MANAGER_REVIEW

    async with session_factory() as session:
        task = await TaskRepository(session).get(task_id)
    assert task is not None and task.project_id == project_id


async def test_running_a_direct_task_is_refused_with_409(client, session_factory, seeded):
    manager = await signup(client, "direct-run@example.com")
    headers = auth_header(manager["access_token"])
    body = await _wizard_org(client, manager["access_token"], "سازمان ب", "direct-run")
    org_id = body["organization"]["id"]
    created = await client.post(
        f"/api/organizations/{org_id}/tasks", json={"title": "کار مستقیم"}, headers=headers
    )
    assert created.status_code == 201, created.text
    task_id = created.json()["id"]

    run = await client.post(
        f"/api/organizations/{org_id}/tasks/{task_id}/runs", json={}, headers=headers
    )
    assert run.status_code == 409, run.text
    assert ENGINE_OWNS in run.json()["error"]["message"]

    # Nothing ran outside the engine: the task still has no run.
    detail = await client.get(f"/api/organizations/{org_id}/tasks/{task_id}", headers=headers)
    assert detail.status_code == 200, detail.text
    assert detail.json()["runs"] == []


async def test_an_mcp_task_enters_the_engine(session_factory, seeded):
    org_id, user_id = await _org_with_owner(session_factory, "direct-mcp")
    token = await _mcp_token(session_factory, org_id, user_id)
    gateway = McpGateway(
        session_factory=session_factory, rate_limiter=TokenRateLimiter(limit_per_minute=None)
    )

    created = await gateway.call(
        tool="tasks_create", arguments={"title": "کار از راه MCP"}, token=token
    )
    assert created.ok is True, created.error
    task_id = created.payload["task"]["id"]

    state = await _state(session_factory, task_id)
    assert state is not None, "an MCP task must get a pipeline state too"
    assert state.organization_id == org_id
    assert state.stage is Stage.PENDING_MANAGER_REVIEW


async def test_a_pre_v2_task_still_runs_through_the_legacy_path(client, session_factory, seeded):
    manager = await signup(client, "direct-legacy@example.com")
    headers = auth_header(manager["access_token"])
    body = await _wizard_org(client, manager["access_token"], "سازمان پ", "direct-legacy")
    org_id = body["organization"]["id"]
    task_id = await legacy_task(session_factory, org_id, manager["user"]["id"], title="کار قدیمی")
    assert await _state(session_factory, task_id) is None

    run = await client.post(
        f"/api/organizations/{org_id}/tasks/{task_id}/runs", json={}, headers=headers
    )
    assert run.status_code == 201, run.text
    assert run.json()["status"] == "completed", run.text