roboco/tests/integration/test_metrics_instrumentation.py

"""New audit instrumentation feeding the extra per-member metrics (phase 4).

- apply_escalation -> task.escalated (escalations metric)
- _unblock_dependents -> task.unblocked_dependents (blocked-others metric)
- mark_agent_idle -> agent.idle (idle/utilization metric)
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any
from uuid import uuid4

import pytest
import pytest_asyncio
from roboco.db.tables import AgentTable, AuditLogTable, ProjectTable, TaskTable
from roboco.models.base import (
    AgentRole,
    AgentStatus,
    Complexity,
    TaskNature,
    TaskStatus,
    TaskType,
    Team,
)
from roboco.services.task import TaskService
from sqlalchemy import select

if TYPE_CHECKING:
    from collections.abc import AsyncIterator

    from sqlalchemy.ext.asyncio import AsyncSession


def _agent(role: AgentRole, slug: str) -> AgentTable:
    return AgentTable(
        id=uuid4(),
        name=slug,
        slug=slug,
        role=role,
        team=Team.BACKEND,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="x",
        capabilities=[],
        permissions={},
        metrics={},
    )


def _task(project_id: Any, created_by: Any, **over: Any) -> TaskTable:
    base: dict[str, Any] = {
        "id": uuid4(),
        "title": "t",
        "description": "d",
        "acceptance_criteria": ["ac"],
        "task_type": TaskType.CODE,
        "nature": TaskNature.TECHNICAL,
        "status": TaskStatus.IN_PROGRESS,
        "team": Team.BACKEND,
        "project_id": project_id,
        "created_by": created_by,
        "estimated_complexity": Complexity.MEDIUM,
    }
    base.update(over)
    return TaskTable(**base)


async def _audit_of(db: AsyncSession, event_type: str, target_id: Any) -> list[Any]:
    rows = (
        (
            await db.execute(
                select(AuditLogTable).where(
                    AuditLogTable.event_type == event_type,
                    AuditLogTable.target_id == target_id,
                )
            )
        )
        .scalars()
        .all()
    )
    return list(rows)


@pytest_asyncio.fixture
async def env(db_session: AsyncSession) -> AsyncIterator[dict]:
    dev = _agent(AgentRole.DEVELOPER, f"be-dev-{uuid4().hex[:6]}")
    pm = _agent(AgentRole.CELL_PM, f"be-pm-{uuid4().hex[:6]}")
    db_session.add_all([dev, pm])
    await db_session.flush()
    project = ProjectTable(
        id=uuid4(),
        name="P",
        slug=f"p-{uuid4().hex[:6]}",
        git_url="https://example.com/r.git",
        assigned_cell=Team.BACKEND,
        created_by=dev.id,
    )
    db_session.add(project)
    await db_session.flush()
    yield {
        "svc": TaskService(db_session),
        "db": db_session,
        "project_id": project.id,
        "dev": dev,
        "pm": pm,
    }


@pytest.mark.asyncio
async def test_apply_escalation_emits_task_escalated(env: dict) -> None:
    db = env["db"]
    task = _task(
        env["project_id"],
        env["dev"].id,
        assigned_to=env["dev"].id,
        claimed_by=env["dev"].id,
    )
    db.add(task)
    await db.flush()
    ok = await env["svc"].apply_escalation(
        task=task,
        target_agent_id=env["pm"].id,
        escalator_slug=env["dev"].slug,
        target_slug=env["pm"].slug,
        reason="need help with the seam",
    )
    assert ok is True
    rows = await _audit_of(db, "task.escalated", task.id)
    assert len(rows) == 1
    assert rows[0].details["escalator_slug"] == env["dev"].slug


@pytest.mark.asyncio
async def test_unblock_dependents_emits_count(env: dict) -> None:
    db = env["db"]
    blocker = _task(env["project_id"], env["dev"].id, status=TaskStatus.COMPLETED)
    db.add(blocker)
    await db.flush()
    dependent = _task(
        env["project_id"],
        env["dev"].id,
        status=TaskStatus.BLOCKED,
        dependency_ids=[blocker.id],
        assigned_to=env["dev"].id,
        claimed_by=env["dev"].id,
    )
    db.add(dependent)
    await db.flush()
    await env["svc"]._unblock_dependents(blocker.id)
    rows = await _audit_of(db, "task.unblocked_dependents", blocker.id)
    assert len(rows) == 1
    assert rows[0].details["count"] == 1


@pytest.mark.asyncio
async def test_mark_agent_idle_emits_agent_idle(env: dict) -> None:
    db = env["db"]
    await env["svc"].mark_agent_idle(env["dev"].id)
    rows = await _audit_of(db, "agent.idle", env["dev"].id)
    assert len(rows) == 1
    assert rows[0].details["agent_slug"] == env["dev"].slug
    refreshed = await db.get(AgentTable, env["dev"].id)
    assert refreshed is not None
    assert refreshed.status == AgentStatus.IDLE