External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix: open findings cleanup (#122)

Browse Source 
* refactor(usage): remove the unconsumed per-agent USAGE_UPDATE event

USAGE_UPDATE was published per active agent each sweep, bridged, and
broadcast to /ws/system, but no panel client ever consumed it — the
dashboard reads only the aggregate USAGE_SNAPSHOT. Every emission was
wasted event-bus and WebSocket traffic.

Drop the UsageUpdate payload, publish_usage_update and its throttle, the
EventType member, and the bridge subscription. Keep USAGE_SNAPSHOT, which
already carries the per-agent breakdown, so no live data is lost.

* refactor(prompter): remove the legacy local-LLM HTTP endpoints

The panel uses only the live SDK-intake path (/prompter/live/*); the legacy
/prompter/chat, /draft and /sessions/* endpoints — backed by the local Ollama
LLM with hardcoded prompts — had no remaining caller. Remove the router, its
mount in app.py, and its integration test. The live router and the shared
draft-confirmation service are untouched.

* refactor(prompter): drop the dead legacy local-LLM service + schemas

With the legacy HTTP endpoints gone, the local-LLM chat/draft/session methods,
their prompt constants, the ConfirmOverrides/TurnResult dataclasses, and the
entire prompter schema module had no production caller (only their own tests).
Remove them, keeping the live-intake path: create_task_from_draft /
confirm_live_draft, the enum/priority/team coercion, and the pure
description/readiness helpers.

* refactor(agents): stop granting the Task sub-agent tool to roles

Every agent role was granted the built-in Task tool, but no role prompt or
workflow uses it and there are no custom sub-agent definitions — so a Task call
only spawns a context-blind generic sub-agent that burns budget (ToolSearch,
the comment's stated use, is MCP-only and not callable in agent containers).

Drop Task from all three grant points in lockstep: the --tools spawn flag and
both _ROLE_BUILTIN_TOOLS maps (system-prompt + briefing layers), with a
regression guard added to each layer's test.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
2026-06-12 23:10:21 +02:00
committed by GitHub
co-authored by Renn F
parent cdde3b256e
commit 718d7dd83e
16 changed files with 84 additions and 2941 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_prompter_routes.py
-862
View File
@@ -1,862 +0,0 @@
"""Prompter API route integration tests.
Covers both the new session-based endpoints:
POST /sessions, POST /sessions/{id}/messages, GET /sessions/{id}/draft,
POST /sessions/{id}/confirm
And the legacy stateless endpoints:
POST /chat, POST /draft
"""
from __future__ import annotations
import json
from http import HTTPStatus
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, patch
from uuid import uuid4
import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from roboco.api.deps import get_agent_context, get_db
from roboco.api.routes.prompter import router as prompter_router
from roboco.db.tables import AgentTable, ProjectTable
from roboco.models.base import AgentRole, AgentStatus, Team
from roboco.models.permissions import AgentContext
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession
# Expected message counts in multi-turn tests
_SINGLE_TURN_MSGS = 2 # 1 user + 1 assistant
_DOUBLE_TURN_MSGS = 4 # 2 user + 2 assistant
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest_asyncio.fixture
async def prompter_client(
db_session: AsyncSession,
) -> AsyncIterator[dict[str, Any]]:
agent_id = uuid4()
agent = AgentTable(
id=agent_id,
name="DevAgent",
slug=f"dev-agent-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(agent)
await db_session.flush()
app = FastAPI()
app.include_router(prompter_router, prefix="/api/prompter")
async def _override_db() -> AsyncIterator[AsyncSession]:
yield db_session
async def _override_agent() -> AgentContext:
return AgentContext(
agent_id=agent_id,
role=AgentRole.DEVELOPER,
team=None,
)
app.dependency_overrides[get_db] = _override_db
app.dependency_overrides[get_agent_context] = _override_agent
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "agent": agent, "db": db_session}
app.dependency_overrides.clear()
@pytest_asyncio.fixture
async def project_fixture(db_session: AsyncSession) -> ProjectTable:
"""Create a minimal project for task creation in confirm tests."""
creator = AgentTable(
id=uuid4(),
name="ProjectCreator",
slug=f"proj-creator-{uuid4().hex[:8]}",
role=AgentRole.MAIN_PM,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(creator)
await db_session.flush()
project = ProjectTable(
id=uuid4(),
name="Test Project",
slug=f"test-project-{uuid4().hex[:8]}",
git_url="https://github.com/test/repo.git",
default_branch="main",
assigned_cell=Team.BACKEND,
created_by=creator.id,
)
db_session.add(project)
await db_session.flush()
return project
_HDR = {"X-Agent-ID": "be-dev-1", "X-Agent-Role": "developer"}
@pytest_asyncio.fixture
async def cross_request_client(
_test_database_url: str,
) -> AsyncIterator[dict[str, Any]]:
"""Client whose DB dependency yields a fresh, NON-auto-committing session
per request.
This is the boundary the shared-session ``prompter_client`` fixture can't
exercise: here a write is only visible to the next request if the route
committed it explicitly. The seed agent is committed up front so both
requests can resolve it.
"""
engine = create_async_engine(_test_database_url, future=True, pool_pre_ping=True)
maker = async_sessionmaker(bind=engine, expire_on_commit=False)
agent_id = uuid4()
async with maker() as seed:
seed.add(
AgentTable(
id=agent_id,
name="XReqAgent",
slug=f"xreq-agent-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
)
await seed.commit()
app = FastAPI()
app.include_router(prompter_router, prefix="/api/prompter")
async def _override_db() -> AsyncIterator[AsyncSession]:
# A fresh session per request that does NOT commit on teardown, so
# persistence depends solely on the route's explicit commit.
async with maker() as session:
yield session
async def _override_agent() -> AgentContext:
return AgentContext(agent_id=agent_id, role=AgentRole.DEVELOPER, team=None)
app.dependency_overrides[get_db] = _override_db
app.dependency_overrides[get_agent_context] = _override_agent
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield {"client": client, "agent_id": agent_id}
app.dependency_overrides.clear()
await engine.dispose()
@pytest.mark.asyncio
async def test_session_persists_across_requests(cross_request_client: dict) -> None:
"""A created session must survive into the next request's own DB session.
Regression for the production 404: the create returned 201 but the session
write was never committed, so the immediately-following /messages call could
not find it. Without the route's explicit commit, this is a 404.
"""
client = cross_request_client["client"]
create = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
assert create.status_code == HTTPStatus.CREATED
session_id = create.json()["id"]
reply = (
'ack\n```roboco-meta\n{"covered": [], "ready": false, "scale": "single"}\n```'
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=reply,
):
msg = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "hello"},
headers=_HDR,
)
assert msg.status_code == HTTPStatus.OK, msg.json()
assert len(msg.json()["messages"]) == _SINGLE_TURN_MSGS
# =============================================================================
# Session-based endpoint tests
# =============================================================================
@pytest.mark.asyncio
async def test_create_session_success(prompter_client: dict) -> None:
"""POST /sessions creates a new session linked to the agent."""
client = prompter_client["client"]
response = await client.post(
"/api/prompter/sessions",
json={},
headers=_HDR,
)
assert response.status_code == HTTPStatus.CREATED
body = response.json()
assert "id" in body
assert body["status"] == "active"
assert "agent_id" in body
assert "created_at" in body
@pytest.mark.asyncio
async def test_send_message_success(prompter_client: dict) -> None:
"""POST /sessions/{id}/messages appends user+assistant messages."""
client = prompter_client["client"]
# Create session
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
mock_response = "Great! Let's gather requirements."
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a new feature"},
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
messages = body["messages"]
assert len(messages) == _SINGLE_TURN_MSGS
roles = [m["role"] for m in messages]
assert "user" in roles
assert "assistant" in roles
assert messages[-1]["content"] == "Great! Let's gather requirements."
assert body["draft_ready"] is False
@pytest.mark.asyncio
async def test_send_message_marks_draft_ready(prompter_client: dict) -> None:
"""A ready roboco-meta control block flips draft_ready and session status."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
mock_response = (
"Understood — I have what I need.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "Add a login page with MFA support"},
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
assert len(body["messages"]) == _SINGLE_TURN_MSGS
assert body["draft_ready"] is True
assert body["scale"] == "single"
# The control block must not leak into the persisted assistant message.
assert "roboco-meta" not in body["messages"][-1]["content"]
@pytest.mark.asyncio
async def test_send_message_not_found(prompter_client: dict) -> None:
"""POST /sessions/{id}/messages with unknown session → 404."""
client = prompter_client["client"]
response = await client.post(
f"/api/prompter/sessions/{uuid4()}/messages",
json={"content": "Hello"},
headers=_HDR,
)
assert response.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_get_draft_generates_from_conversation(prompter_client: dict) -> None:
"""GET /sessions/{id}/draft generates a draft via LLM."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
draft_json = {
"title": "Add login page",
"description": "Implement a secure login page with email and password",
"acceptance_criteria": [
"User can enter email and password",
"Invalid credentials show error message",
],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
"priority": 2,
}
chat_response = "Tell me more about the requirements."
draft_response = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_response,
):
await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a login page"},
headers=_HDR,
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=draft_response,
):
response = await client.get(
f"/api/prompter/sessions/{session_id}/draft",
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
assert body["draft"]["title"] == "Add login page"
assert body["draft"]["source"] == "prompter"
assert body["confirmed_at"] is None
assert body["draft"]["confirmed_by_human"] is False
assert body["session_id"] == session_id
@pytest.mark.asyncio
async def test_get_draft_cached(prompter_client: dict) -> None:
"""GET /sessions/{id}/draft returns the cached draft on subsequent calls."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
draft_json = {
"title": "Add login page",
"description": "Implement a secure login page with email and password",
"acceptance_criteria": ["User can enter credentials"],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
"priority": 2,
}
chat_response = "Got it."
draft_response = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_response,
):
await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a login page"},
headers=_HDR,
)
call_count = 0
async def _mock_create(**_kwargs: Any) -> str:
nonlocal call_count
call_count += 1
return draft_response
with patch(
"roboco.services.prompter.PrompterService._create_message",
side_effect=_mock_create,
):
await client.get(f"/api/prompter/sessions/{session_id}/draft", headers=_HDR)
second_response = await client.get(
f"/api/prompter/sessions/{session_id}/draft", headers=_HDR
)
assert second_response.status_code == HTTPStatus.OK
# LLM should only be called once (draft is cached)
assert call_count == 1
@pytest.mark.asyncio
async def test_get_draft_empty_session_returns_400(prompter_client: dict) -> None:
"""GET /sessions/{id}/draft with no messages → 400."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
response = await client.get(
f"/api/prompter/sessions/{session_id}/draft",
headers=_HDR,
)
assert response.status_code == HTTPStatus.BAD_REQUEST
@pytest.mark.asyncio
async def test_confirm_draft_creates_task(
prompter_client: dict, project_fixture: ProjectTable
) -> None:
"""POST /sessions/{id}/confirm validates draft and creates a real task."""
client = prompter_client["client"]
project_id = str(project_fixture.id)
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
draft_json = {
"title": "Add login page",
"description": "Implement a secure login page with email and password",
"acceptance_criteria": ["User can enter credentials"],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
"priority": 2,
}
chat_response = "Got it."
draft_response = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_response,
):
await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a login page"},
headers=_HDR,
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=draft_response,
):
await client.get(f"/api/prompter/sessions/{session_id}/draft", headers=_HDR)
confirm_response = await client.post(
f"/api/prompter/sessions/{session_id}/confirm",
json={"project_id": project_id},
headers=_HDR,
)
assert confirm_response.status_code == HTTPStatus.CREATED
body = confirm_response.json()
assert "task_id" in body
assert body["task_id"] is not None
@pytest.mark.asyncio
async def test_confirm_draft_requires_project_or_product(
prompter_client: dict,
) -> None:
"""POST /sessions/{id}/confirm without project_id/product_id → 400."""
client = prompter_client["client"]
session_resp = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
session_id = session_resp.json()["id"]
draft_json = {
"title": "Add login page",
"description": "Implement a secure login page with email and password",
"acceptance_criteria": ["User can enter credentials"],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
"priority": 2,
}
chat_response = "Got it."
draft_response = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_response,
):
await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a login page"},
headers=_HDR,
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=draft_response,
):
await client.get(f"/api/prompter/sessions/{session_id}/draft", headers=_HDR)
confirm_response = await client.post(
f"/api/prompter/sessions/{session_id}/confirm",
json={},
headers=_HDR,
)
assert confirm_response.status_code == HTTPStatus.BAD_REQUEST
# =============================================================================
# Full happy path integration test
# =============================================================================
@pytest.mark.asyncio
async def test_full_happy_path(
prompter_client: dict, project_fixture: ProjectTable
) -> None:
"""Full happy path: create session → send messages → get draft → confirm task."""
client = prompter_client["client"]
project_id = str(project_fixture.id)
# Step 1: Create session
step1 = await client.post("/api/prompter/sessions", json={}, headers=_HDR)
assert step1.status_code == HTTPStatus.CREATED
session_id = step1.json()["id"]
# Step 2: Send messages
chat_mock = "Please describe the acceptance criteria for this feature."
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_mock,
):
step2a = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "I need a dark mode toggle for the UI"},
headers=_HDR,
)
assert step2a.status_code == HTTPStatus.OK
chat_mock2 = "I have enough information to draft a task now."
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=chat_mock2,
):
step2b = await client.post(
f"/api/prompter/sessions/{session_id}/messages",
json={"content": "Preference is persisted across sessions"},
headers=_HDR,
)
assert step2b.status_code == HTTPStatus.OK
messages = step2b.json()["messages"]
assert len(messages) == _DOUBLE_TURN_MSGS
# Step 3: Get draft
draft_json = {
"title": "Add dark mode toggle",
"description": "Implement a dark mode toggle so users can switch themes",
"acceptance_criteria": [
"User can toggle light/dark mode",
"Preference is persisted across sessions",
],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "low",
"priority": 2,
}
draft_mock = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=draft_mock,
):
step3 = await client.get(
f"/api/prompter/sessions/{session_id}/draft", headers=_HDR
)
assert step3.status_code == HTTPStatus.OK
draft_body = step3.json()
assert draft_body["draft"]["title"] == "Add dark mode toggle"
# Step 4: Confirm draft → creates task
step4 = await client.post(
f"/api/prompter/sessions/{session_id}/confirm",
json={"project_id": project_id},
headers=_HDR,
)
assert step4.status_code == HTTPStatus.CREATED
task_body = step4.json()
assert "task_id" in task_body
assert task_body["task_id"] is not None
# =============================================================================
# Legacy stateless endpoint tests (backward compatibility)
# =============================================================================
@pytest.mark.asyncio
async def test_prompter_chat_success(prompter_client: dict) -> None:
client = prompter_client["client"]
mock_response = "Great! Let's gather requirements."
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
"/api/prompter/chat",
json={
"messages": [{"role": "user", "content": "I need a new feature"}],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
assert body["message"] == "Great! Let's gather requirements."
assert body["draft_ready"] is False
@pytest.mark.asyncio
async def test_prompter_chat_draft_ready(prompter_client: dict) -> None:
client = prompter_client["client"]
mock_response = (
"Understood.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
"/api/prompter/chat",
json={
"messages": [
{"role": "user", "content": "I need a new feature"},
{"role": "assistant", "content": "Tell me more"},
{"role": "user", "content": "Add a login page"},
],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
assert body["draft_ready"] is True
@pytest.mark.asyncio
async def test_prompter_chat_llm_failure(prompter_client: dict) -> None:
client = prompter_client["client"]
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
side_effect=Exception("Anthropic API unavailable"),
):
response = await client.post(
"/api/prompter/chat",
json={
"messages": [{"role": "user", "content": "Hello"}],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
body = response.json()
assert "LLM chat failed" in body["detail"]["message"]
@pytest.mark.asyncio
async def test_prompter_draft_success(prompter_client: dict) -> None:
client = prompter_client["client"]
draft_json = {
"title": "Add login page",
"description": "Implement a secure login page with email and password",
"acceptance_criteria": [
"User can enter email and password",
"Invalid credentials show error message",
],
"team": "frontend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
"priority": 2,
}
mock_response = json.dumps(draft_json)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
"/api/prompter/draft",
json={
"messages": [
{"role": "user", "content": "I need a login page"},
],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.OK
body = response.json()
assert body["draft"]["title"] == "Add login page"
assert body["draft"]["source"] == "prompter"
assert body["draft"]["confirmed_by_human"] is False
assert "reasoning" in body
@pytest.mark.asyncio
async def test_prompter_draft_invalid_json_from_llm(prompter_client: dict) -> None:
client = prompter_client["client"]
mock_response = "not valid json"
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
"/api/prompter/draft",
json={
"messages": [{"role": "user", "content": "Hello"}],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.BAD_REQUEST
body = response.json()
assert "Draft response was not valid JSON" in body["detail"]["message"]
@pytest.mark.asyncio
async def test_prompter_draft_schema_mismatch(prompter_client: dict) -> None:
client = prompter_client["client"]
bad_draft = {
"title": "x",
"description": "too short",
}
mock_response = json.dumps(bad_draft)
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
return_value=mock_response,
):
response = await client.post(
"/api/prompter/draft",
json={
"messages": [{"role": "user", "content": "Hello"}],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
body = response.json()
assert "draft_schema_error" in body["detail"]["error"]
@pytest.mark.asyncio
async def test_prompter_draft_llm_failure(prompter_client: dict) -> None:
client = prompter_client["client"]
with patch(
"roboco.services.prompter.PrompterService._create_message",
new_callable=AsyncMock,
side_effect=Exception("Anthropic API unavailable"),
):
response = await client.post(
"/api/prompter/draft",
json={
"messages": [{"role": "user", "content": "Hello"}],
},
headers=_HDR,
)
assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
body = response.json()
assert "LLM draft generation failed" in body["detail"]["message"]
@pytest.mark.asyncio
async def test_prompter_chat_empty_messages(prompter_client: dict) -> None:
client = prompter_client["client"]
response = await client.post(
"/api/prompter/chat",
json={"messages": []},
headers=_HDR,
)
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_prompter_chat_invalid_role(prompter_client: dict) -> None:
client = prompter_client["client"]
response = await client.post(
"/api/prompter/chat",
json={"messages": [{"role": "invalid", "content": "hi"}]},
headers=_HDR,
)
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_prompter_draft_empty_messages(prompter_client: dict) -> None:
client = prompter_client["client"]
response = await client.post(
"/api/prompter/draft",
json={"messages": []},
headers=_HDR,
)
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
tests/unit/agents/test_tool_load_directive_layer.py
+18
View File
@@ -95,6 +95,24 @@ def test_pm_blocks_exclude_edit_and_write() -> None:
)
def test_no_role_is_granted_the_task_subagent_tool() -> None:
"""Task (sub-agent dispatch) is dropped from the built-in tool grant.
No role prompt or workflow uses Task and there are no custom sub-agent
definitions, so a Task call only spawns a context-blind generic sub-agent
that burns budget. The tools-ready line must not advertise it for any role.
"""
for role in (
AgentRole.DEVELOPER,
AgentRole.DOCUMENTER,
AgentRole.QA,
AgentRole.MAIN_PM,
AgentRole.CELL_PM,
):
names = _tool_names(_composed_prompt_for(role, Team.BACKEND))
assert "Task" not in names, f"{role.value} must not list Task: {names}"
def test_block_is_first_layer_before_lifecycle() -> None:
"""Tools-ready block precedes the lifecycle and base layers."""
prompt = _composed_prompt_for(AgentRole.DEVELOPER, Team.BACKEND)
tests/unit/api/test_schemas_prompter.py
-272
View File
@@ -1,272 +0,0 @@
"""Unit tests for Prompter API schemas.
Covers schema validation for both the session-based and legacy schemas.
"""
from __future__ import annotations
from uuid import uuid4
import pytest
from pydantic import ValidationError as PydanticValidationError
from roboco.api.schemas.prompter import (
CellWork,
ChatMessage,
PrompterChatRequest,
PrompterDraftTask,
PrompterMessageRequest,
PrompterTurnResponse,
TaskConfirmRequest,
)
# =============================================================================
# ChatMessage
# =============================================================================
def test_chat_message_valid_roles() -> None:
for role in ("user", "assistant", "system"):
msg = ChatMessage(role=role, content="Hello")
assert msg.role == role
def test_chat_message_invalid_role() -> None:
with pytest.raises(PydanticValidationError) as exc_info:
ChatMessage(role="admin", content="Hello")
assert "role must be one of" in str(exc_info.value)
def test_chat_message_empty_content() -> None:
with pytest.raises(PydanticValidationError):
ChatMessage(role="user", content="")
# =============================================================================
# PrompterMessageRequest
# =============================================================================
def test_message_request_valid() -> None:
req = PrompterMessageRequest(content="I need a feature")
assert req.content == "I need a feature"
assert req.context == {}
def test_message_request_empty_content() -> None:
with pytest.raises(PydanticValidationError):
PrompterMessageRequest(content="")
def test_message_request_with_context() -> None:
req = PrompterMessageRequest(content="Hello", context={"key": "value"})
assert req.context["key"] == "value"
# =============================================================================
# TaskConfirmRequest
# =============================================================================
def test_task_confirm_request_all_optional() -> None:
req = TaskConfirmRequest()
assert req.project_id is None
assert req.product_id is None
assert req.assigned_to is None
assert req.overrides == {}
def test_task_confirm_request_with_project() -> None:
pid = uuid4()
req = TaskConfirmRequest(project_id=pid)
assert req.project_id == pid
# =============================================================================
# PrompterDraftTask
# =============================================================================
def test_draft_task_valid() -> None:
draft = PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
assert draft.title == "Add login page"
assert draft.source == "prompter"
assert draft.confirmed_by_human is False
def test_draft_task_title_too_long() -> None:
with pytest.raises(PydanticValidationError):
PrompterDraftTask(
title="x" * 201,
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
def test_draft_task_description_too_short() -> None:
with pytest.raises(PydanticValidationError):
PrompterDraftTask(
title="Add login page",
description="short", # <20 chars
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
def test_draft_task_empty_acceptance_criteria() -> None:
with pytest.raises(PydanticValidationError):
PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=[],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
def test_draft_task_invalid_team() -> None:
with pytest.raises(PydanticValidationError):
PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="infra", # invalid
task_type="code",
nature="technical",
estimated_complexity="medium",
)
def test_draft_task_priority_bounds() -> None:
# Valid bounds
for p in (0, 1, 2, 3):
d = PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
priority=p,
)
assert d.priority == p
# Out of bounds
with pytest.raises(PydanticValidationError):
PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
priority=4,
)
# =============================================================================
# Structured spec fields
# =============================================================================
def test_cell_work_valid() -> None:
cw = CellWork(team="backend", summary="Build the endpoint", items=["Route", "Test"])
assert cw.team.value == "backend"
assert cw.items == ["Route", "Test"]
def test_cell_work_requires_summary() -> None:
with pytest.raises(PydanticValidationError):
CellWork(team="backend", summary="")
def test_draft_task_structured_fields_default_empty() -> None:
draft = PrompterDraftTask(
title="Add login page",
description="Implement a secure login page with email and password",
acceptance_criteria=["User can log in"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="medium",
)
assert draft.objective is None
assert draft.what_this_builds == []
assert draft.the_work == []
assert draft.notes == []
def test_draft_task_with_structured_fields() -> None:
draft = PrompterDraftTask(
title="Ship the Prompter",
description="A board-led feature spanning three cells, fully wired.",
acceptance_criteria=["It works end to end"],
team="backend",
task_type="code",
nature="technical",
estimated_complexity="high",
objective="Let humans chat a task into existence.",
what_this_builds=["A /prompter page", "A chat endpoint"],
the_work=[
CellWork(team="backend", summary="Chat endpoint", items=["Route"]),
CellWork(team="frontend", summary="Chat UI", items=["Page"]),
],
notes=["Reuse the LLM service"],
)
assert [w.team.value for w in draft.the_work] == ["backend", "frontend"]
def test_confirm_request_carries_edited_draft() -> None:
draft = PrompterDraftTask(
title="Edited title",
description="An edited description that clears the minimum length.",
acceptance_criteria=["Done"],
team="frontend",
task_type="code",
nature="technical",
estimated_complexity="low",
)
req = TaskConfirmRequest(project_id=uuid4(), draft=draft)
assert req.draft is not None
assert req.draft.title == "Edited title"
def test_turn_response_shape() -> None:
resp = PrompterTurnResponse(messages=[], draft_ready=True, scale="multi")
assert resp.draft_ready is True
assert resp.scale == "multi"
assert resp.messages == []
# =============================================================================
# PrompterChatRequest (legacy)
# =============================================================================
def test_chat_request_requires_messages() -> None:
with pytest.raises(PydanticValidationError):
PrompterChatRequest(messages=[])
def test_chat_request_valid() -> None:
req = PrompterChatRequest(messages=[ChatMessage(role="user", content="Hello")])
assert len(req.messages) == 1
assert req.context == {}
tests/unit/api/test_websocket_bridge.py
+1 -29
View File
@@ -297,33 +297,6 @@ async def test_handle_rate_limit_ignores_unrelated_event() -> None:
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_handle_usage_update_broadcasts_to_system() -> None:
"""USAGE_UPDATE event → broadcast_system tagged USAGE_UPDATE + data fields."""
expected_input = 100
expected_output = 50
event = _evt(
EventType.USAGE_UPDATE,
{
"agent_id": "be-dev-1",
"task_id": "task-abc",
"input_tokens": expected_input,
"output_tokens": expected_output,
"model": "claude-sonnet-4-6",
"timestamp": "2026-06-11T00:00:00+00:00",
},
)
with patch("roboco.api.websocket_bridge.manager") as mgr:
mgr.broadcast_system = AsyncMock()
await _handle_usage_event(event)
mgr.broadcast_system.assert_awaited_once()
msg = mgr.broadcast_system.await_args.args[0]
assert msg["type"] == "USAGE_UPDATE"
assert msg["agent_id"] == "be-dev-1"
assert msg["input_tokens"] == expected_input
assert msg["output_tokens"] == expected_output
@pytest.mark.asyncio
async def test_handle_usage_snapshot_broadcasts_to_system() -> None:
"""USAGE_SNAPSHOT event → broadcast_system tagged USAGE_SNAPSHOT + aggregate."""
@@ -376,7 +349,7 @@ def test_register_websocket_bridge_handlers_subscribes_all_event_types() -> None
with patch("roboco.api.websocket_bridge.get_event_bus", return_value=fake):
register_websocket_bridge_handlers()
types = [t for t, _ in fake.subscribed]
# All 14 expected event types appear at least once.
# All expected event types appear at least once.
assert EventType.NOTIFICATION_SENT in types
assert EventType.NOTIFICATION_ACKED in types
assert EventType.SESSION_CREATED in types
@@ -390,7 +363,6 @@ def test_register_websocket_bridge_handlers_subscribes_all_event_types() -> None
assert EventType.RATE_LIMIT_HIT in types
assert EventType.RATE_LIMIT_LIFTED in types
# Usage events forwarded to /ws/system.
assert EventType.USAGE_UPDATE in types
assert EventType.USAGE_SNAPSHOT in types
tests/unit/runtime/test_briefing_tool_load_block.py
+20
View File
@@ -68,6 +68,26 @@ def test_pm_blocks_exclude_edit_and_write() -> None:
assert "Write" not in names, f"{role} must not list Write"
def test_no_role_block_lists_the_task_subagent_tool() -> None:
"""Task (sub-agent dispatch) is dropped from the briefing tool grant.
No role uses Task and there are no custom sub-agent definitions, so it only
spawns a context-blind generic sub-agent that burns budget.
"""
for role in (
"developer",
"documenter",
"qa",
"main_pm",
"cell_pm",
"product_owner",
"head_marketing",
"auditor",
):
names = _tool_names(_orch()._build_tool_load_block(role))
assert "Task" not in names, f"{role} must not list Task: {names}"
def test_unknown_role_returns_empty() -> None:
assert _orch()._build_tool_load_block("nonexistent") == ""
tests/unit/services/test_prompter.py
+6 -251
View File
@@ -1,14 +1,14 @@
"""Unit tests for PrompterService.
Tests the service layer logic with mocked LLM calls. Uses an in-memory
async session (via conftest fixtures) for DB-backed tests.
Covers the live-intake draft → task flow (``create_task_from_draft`` /
``confirm_live_draft`` + the enum/priority/team coercion) and the pure
draft/description helpers. DB-backed tests use an in-memory async session via
conftest fixtures.
"""
from __future__ import annotations
import json
from typing import Any
from unittest.mock import AsyncMock, patch
from uuid import UUID, uuid4
import pytest
@@ -28,12 +28,9 @@ from roboco.models.base import (
Team,
)
from roboco.seeds.initial_data import AGENT_UUIDS
from roboco.services.base import NotFoundError, ServiceError, ValidationError
from roboco.services.base import ServiceError
from roboco.services.prompter import (
PrompterService,
_build_chat_prompt,
_build_draft_prompt,
_build_reasoning,
compose_description,
derive_scale,
get_prompter_service,
@@ -209,41 +206,6 @@ def test_coerce_draft_enums_keeps_valid_and_derives_missing_team() -> None:
assert complexity is Complexity.MEDIUM
def test_build_chat_prompt_basic() -> None:
messages = [
{"role": "user", "content": "I need a feature"},
{"role": "assistant", "content": "Tell me more"},
]
prompt = _build_chat_prompt(messages, None)
assert "user: I need a feature" in prompt
assert "assistant: Tell me more" in prompt
assert "Continue the conversation" in prompt
def test_build_chat_prompt_with_context() -> None:
messages = [{"role": "user", "content": "hello"}]
prompt = _build_chat_prompt(messages, {"team": "backend"})
assert "Context:" in prompt
assert "team: backend" in prompt
def test_build_draft_prompt() -> None:
messages = [{"role": "user", "content": "I need a login page"}]
prompt = _build_draft_prompt(messages, None)
assert "valid JSON" in prompt
assert "user: I need a login page" in prompt
def test_build_reasoning() -> None:
messages = [{"role": "user", "content": "Hello"}] * 3
draft = {"title": "My Task", "team": "backend", "estimated_complexity": "medium"}
reasoning = _build_reasoning(messages, draft)
assert "My Task" in reasoning
assert "backend" in reasoning
assert "medium" in reasoning
assert "3 messages" in reasoning
# =============================================================================
# Factory
# =============================================================================
@@ -262,178 +224,10 @@ def test_get_prompter_service_raises_without_db_for_session_methods() -> None:
# =============================================================================
# Stateless chat / draft (with mocked LLM)
# DB-backed: assignee routing + confirm_live_draft
# =============================================================================
@pytest.mark.asyncio
async def test_chat_success_with_mock_llm() -> None:
service = get_prompter_service()
with patch.object(
service,
"_create_message",
new_callable=AsyncMock,
return_value="Great, let's continue!",
):
result = await service.chat(
messages=[{"role": "user", "content": "I need a feature"}]
)
assert result["message"] == "Great, let's continue!"
assert result["draft_ready"] is False
@pytest.mark.asyncio
async def test_chat_draft_ready_signal() -> None:
service = get_prompter_service()
reply = (
"Got it — I have what I need.\n\n"
'```roboco-meta\n{"covered": ["objective", "scope", "surface", '
'"acceptance"], "ready": true, "scale": "single"}\n```'
)
with patch.object(
service,
"_create_message",
new_callable=AsyncMock,
return_value=reply,
):
result = await service.chat(
messages=[{"role": "user", "content": "I need a feature"}]
)
assert result["draft_ready"] is True
assert result["scale"] == "single"
# The control block is stripped from the user-visible reply.
assert "roboco-meta" not in result["message"]
assert result["message"] == "Got it — I have what I need."
@pytest.mark.asyncio
async def test_chat_raises_on_empty_response() -> None:
service = get_prompter_service()
with (
patch.object(
service, "_create_message", new_callable=AsyncMock, return_value=""
),
pytest.raises(ServiceError, match="LLM returned empty content"),
):
await service.chat(messages=[{"role": "user", "content": "Hello"}])
@pytest.mark.asyncio
async def test_chat_raises_on_llm_error() -> None:
service = get_prompter_service()
with (
patch.object(
service,
"_create_message",
new_callable=AsyncMock,
side_effect=Exception("API unavailable"),
),
pytest.raises(ServiceError, match="LLM chat failed"),
):
await service.chat(messages=[{"role": "user", "content": "Hello"}])
@pytest.mark.asyncio
async def test_draft_success_with_mock_llm() -> None:
service = get_prompter_service()
draft_data = {
"title": "Add login",
"description": "Implement login functionality with JWT tokens",
"acceptance_criteria": ["User can log in"],
"team": "backend",
"task_type": "code",
"nature": "technical",
"estimated_complexity": "medium",
}
with patch.object(
service,
"_create_message",
new_callable=AsyncMock,
return_value=json.dumps(draft_data),
):
result = await service.draft(
messages=[{"role": "user", "content": "I need a login feature"}]
)
assert result["draft"]["title"] == "Add login"
assert result["draft"]["source"] == "prompter"
assert result["draft"]["confirmed_by_human"] is False
assert "reasoning" in result
@pytest.mark.asyncio
async def test_draft_raises_on_invalid_json() -> None:
service = get_prompter_service()
with (
patch.object(
service,
"_create_message",
new_callable=AsyncMock,
return_value="Not JSON at all",
),
pytest.raises(ValidationError, match="not valid JSON"),
):
await service.draft(messages=[{"role": "user", "content": "Hello"}])
@pytest.mark.asyncio
async def test_draft_raises_on_llm_error() -> None:
service = get_prompter_service()
with (
patch.object(
service,
"_create_message",
new_callable=AsyncMock,
side_effect=Exception("API unavailable"),
),
pytest.raises(ServiceError, match="LLM draft generation failed"),
):
await service.draft(messages=[{"role": "user", "content": "Hello"}])
# =============================================================================
# Session-based: create_session (DB-backed via conftest)
# =============================================================================
@pytest.mark.asyncio
async def test_create_session_db(db_session: Any) -> None:
"""create_session persists a PrompterSessionTable row."""
service = get_prompter_service(db=db_session)
agent_id = uuid4()
agent = AgentTable(
id=agent_id,
name="TestAgent",
slug=f"test-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(agent)
await db_session.flush()
session = await service.create_session(agent_id=agent_id)
assert session.id is not None
assert session.status == "active"
assert session.agent_id == agent_id
@pytest.mark.asyncio
async def test_assignee_is_board_distinguishes_roles(db_session: Any) -> None:
"""Drives product team routing: a board reviewer keeps the root on the board.
@@ -472,45 +266,6 @@ async def test_assignee_is_board_distinguishes_roles(db_session: Any) -> None:
assert await service._assignee_is_board(uuid4()) is False
@pytest.mark.asyncio
async def test_get_session_not_found(db_session: Any) -> None:
"""_get_session raises NotFoundError for unknown session ID."""
service = get_prompter_service(db=db_session)
with pytest.raises(NotFoundError):
await service._get_session(uuid4(), uuid4())
@pytest.mark.asyncio
async def test_get_draft_empty_session_raises(db_session: Any) -> None:
"""get_or_generate_draft raises ValidationError if no messages exist."""
service = get_prompter_service(db=db_session)
agent_id = uuid4()
agent = AgentTable(
id=agent_id,
name="TestAgent",
slug=f"test-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=None,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(agent)
await db_session.flush()
session = await service.create_session(agent_id=agent_id)
with pytest.raises(ValidationError, match="empty conversation"):
await service.get_or_generate_draft(
session_id=UUID(str(session.id)),
agent_id=agent_id,
)
async def _seed_project_and_ceo(db_session: Any) -> tuple[UUID, UUID]:
"""Seed a system agent + project + CEO; return (project_id, ceo_id).
tests/unit/services/test_usage_events.py
+4 -184
View File
@@ -1,196 +1,16 @@
"""Unit tests for roboco.services.usage_events.
Covers the _UsageThrottle class and the publish_usage_update /
publish_usage_snapshot helpers. No real Redis or event bus is needed —
we use AsyncMock to assert that bus.publish is called with the right
Covers the publish_usage_snapshot helper. No real Redis or event bus is
needed — we use AsyncMock to assert that bus.publish is called with the right
payload and type.
The throttle suppression test is the acceptance-criterion gate:
"Server-side throttle prevents more than 1 USAGE_UPDATE publish per
agent per 5-second window."
"""
from __future__ import annotations
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, MagicMock
import pytest
from roboco.services.usage_events import (
UsageSnapshot,
UsageUpdate,
_UsageThrottle,
publish_usage_snapshot,
publish_usage_update,
)
# ---------------------------------------------------------------------------
# _UsageThrottle
# ---------------------------------------------------------------------------
def test_throttle_allows_first_publish() -> None:
"""A fresh agent has no prior timestamp — first publish is always allowed."""
th = _UsageThrottle(window=5.0)
assert th.should_publish("be-dev-1") is True
def test_throttle_suppresses_second_publish_within_window() -> None:
"""Second call within the 5-second window returns False (suppressed)."""
th = _UsageThrottle(window=5.0)
with patch("roboco.services.usage_events.time") as mock_time:
mock_time.monotonic.return_value = 100.0
assert th.should_publish("be-dev-1") is True # first → allowed
mock_time.monotonic.return_value = 104.9 # 4.9 s later — still inside window
assert th.should_publish("be-dev-1") is False # suppressed
def test_throttle_allows_publish_after_window_expires() -> None:
"""After the full window elapses, the next publish is allowed again."""
th = _UsageThrottle(window=5.0)
with patch("roboco.services.usage_events.time") as mock_time:
mock_time.monotonic.return_value = 100.0
assert th.should_publish("be-dev-1") is True # first
mock_time.monotonic.return_value = 105.0 # exactly 5 s later
assert th.should_publish("be-dev-1") is True # window elapsed → allowed
def test_throttle_tracks_agents_independently() -> None:
"""Different agents have independent throttle windows."""
th = _UsageThrottle(window=5.0)
with patch("roboco.services.usage_events.time") as mock_time:
mock_time.monotonic.return_value = 100.0
assert th.should_publish("be-dev-1") is True
# be-dev-2 has never published, so it is always allowed.
assert th.should_publish("be-dev-2") is True
mock_time.monotonic.return_value = 101.0
# be-dev-1 is suppressed; be-dev-2 is also now suppressed.
assert th.should_publish("be-dev-1") is False
assert th.should_publish("be-dev-2") is False
def test_throttle_records_timestamp_on_allow() -> None:
"""should_publish records the current time when it returns True."""
th = _UsageThrottle(window=5.0)
recorded_at = 200.0
with patch("roboco.services.usage_events.time") as mock_time:
mock_time.monotonic.return_value = recorded_at
th.should_publish("be-dev-1")
assert th._last["be-dev-1"] == recorded_at
# ---------------------------------------------------------------------------
# publish_usage_update
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_publish_usage_update_calls_bus_publish() -> None:
"""First call in a window publishes the event and returns True."""
bus = MagicMock()
bus.publish = AsyncMock()
th = _UsageThrottle(window=5.0)
expected_input = 100
expected_output = 50
with patch("roboco.services.usage_events._throttle", th):
result = await publish_usage_update(
bus,
UsageUpdate(
agent_id="be-dev-1",
task_id="task-abc",
input_tokens=expected_input,
output_tokens=expected_output,
model="claude-sonnet-4-6",
),
)
assert result is True
bus.publish.assert_awaited_once()
event = bus.publish.await_args.args[0]
assert event.type.value == "usage.update"
assert event.data["agent_id"] == "be-dev-1"
assert event.data["task_id"] == "task-abc"
assert event.data["input_tokens"] == expected_input
assert event.data["output_tokens"] == expected_output
assert event.data["model"] == "claude-sonnet-4-6"
assert "timestamp" in event.data
@pytest.mark.asyncio
async def test_publish_usage_update_throttle_suppresses_second_call() -> None:
"""Second publish within the throttle window is suppressed (returns False)."""
bus = MagicMock()
bus.publish = AsyncMock()
th = _UsageThrottle(window=5.0)
with (
patch("roboco.services.usage_events._throttle", th),
patch("roboco.services.usage_events.time") as mock_time,
):
mock_time.monotonic.return_value = 100.0
first = await publish_usage_update(
bus,
UsageUpdate(
agent_id="be-dev-1",
task_id=None,
input_tokens=10,
output_tokens=5,
model="sonnet",
),
)
mock_time.monotonic.return_value = 102.0 # 2 s later — still suppressed
second = await publish_usage_update(
bus,
UsageUpdate(
agent_id="be-dev-1",
task_id=None,
input_tokens=20,
output_tokens=10,
model="sonnet",
),
)
assert first is True
assert second is False
# bus.publish should only have been called once.
assert bus.publish.await_count == 1
@pytest.mark.asyncio
async def test_publish_usage_update_custom_timestamp() -> None:
"""Custom timestamp is passed through to the event data."""
bus = MagicMock()
bus.publish = AsyncMock()
ts = datetime(2026, 6, 11, 12, 0, 0, tzinfo=UTC)
# Use a fresh throttle so the first publish goes through.
th = _UsageThrottle(window=5.0)
with patch("roboco.services.usage_events._throttle", th):
await publish_usage_update(
bus,
UsageUpdate(
agent_id="be-dev-1",
task_id=None,
input_tokens=0,
output_tokens=0,
model="sonnet",
timestamp=ts,
),
)
event = bus.publish.await_args.args[0]
assert event.data["timestamp"] == ts.isoformat()
from roboco.services.usage_events import UsageSnapshot, publish_usage_snapshot
# ---------------------------------------------------------------------------
# publish_usage_snapshot