Files
roboco/tests/integration/test_task_service_misc.py
T
7901ea419e Retire channels/sessions/messages; A2A becomes primary agent comms (#306)
* feat(a2a): deliver latest incoming message preview into the claim briefing

list_unread_a2a now carries last_message_preview (the latest message from the
OTHER agent, never the agent's own reply), fetched via a correlated subquery in
the same query — no N+1 on the per-verb briefing path.

* feat(a2a): read_a2a verb delivers unread message bodies to the agent

A2AService.get_unread_messages returns the caller's unread INCOMING messages
(never its own sends), marking exactly those rows read atomically so a message
arriving mid-call is preserved. Wired as the read_a2a content verb (route +
do_server tool + granted to every delivery role) — the content-bearing read the
A2A inbox lacked (read_messages only zeroed the counter).

* docs(rag): document read_a2a as the A2A content-read path

* fix(task): backlog activation no longer requires a discussion session

Removes the SessionTaskTable gate in activate() (and its dangling log field),
deletes _inherit_parent_session + its create() call, and drops the now-unused
SessionTaskTable import. Coordination rides task state; the session subsystem is
being retired. Tests updated to the new (no-session) behavior.

* fix(orchestrator): drop session sweep from _run_sweep

Removes the messaging import + sweep_timed_out_sessions call. That import sat
outside the try/except, so once messaging.py is deleted it would have killed the
entire sweep cascade (budget kill-switch, token rollups, retention, image prune,
superseded-PR reconcile). Notification sweep + all maintenance sweeps unchanged.

* release-manager --no-tags read-clone fix

* test: update evidence_repo unit test for a2a last_message_preview

* refactor(gateway): drop session propagation on delegate

Removes propagate_sessions_to_subtask from delegate(), the ChoreographerDeps
messaging field + property, and the ChoreographerDeps messaging arg in deps.py
(ContentActions messaging + import stay until the verbs are removed). Deletes the
propagation test; strips the now-invalid messaging kwarg from ChoreographerDeps
test builders.

* refactor(gateway): remove say/open_session/link_session/channels verbs

Removes the four channel/session verbs across content_actions (impls +
ContentActionsDeps.messaging), do_server (tools + registry), role_config (grants
+ _CHANNEL_DISCOVERY), do.py (routes), schemas/v1/do.py (request models), and
deps.py (MessagingService import + construction). Regenerates the prompt verb
tables. dm/notify/read_messages/read_a2a stay. Tests deleted/updated accordingly.

* uv.lock Upgrade

* refactor: remove conversation RAG indexing; Secretary announces via notification

Drops the CONVERSATIONS index (index_conversation, ConversationsIndexPlugin,
IndexType.CONVERSATIONS enum, IndexConversationParams, mentor.py type-label, the
messaging index hook) and its chunk-table manifest entries. The Secretary's
ANNOUNCE/RELAY_MESSAGE now fan out a BROADCAST notification to every agent's
inbox (NotificationService.broadcast) instead of posting to a dead channel.

* fix(panel): label RAG health error lines by subsystem

A red llm_error (e.g. the glm-5.2:cloud weekly-limit 429) rendered under
the 'Embedding: ok' header with no label, reading as an embedding failure.
Prefix each error line with LLM / Embedding / Vector store.

* refactor: remove channel/message reads from metrics, dashboard, git, events

MetricsService drops get_communication_volume + the MessageTable
message-count in get_agent_metrics (and the now-dead messages_sent_week
field). DashboardService drops get_channel_feeds/_compute_channel_status
and the message read in get_recent_activity (task activity kept);
get_auditor_metrics no longer reports communication_volume.
GitService's two primary-session-id helpers always return None now
(callers already treat None as "no primary session"). events/handlers.py
drops the SESSION_CLOSED/SESSION_TIMEOUT subscriptions + the
handle_session_boundary handler.

Forced follow-on: api/routes/dashboard.py + api/schemas/dashboard.py
dropped the now-dangling live_feeds/ChannelFeed surface and the
/metrics/communication route, which wrapped the removed service calls
directly (mypy would otherwise fail on the missing attributes).

* refactor: delete MessagingService + channel seeding

Edited db/__init__.py and services/__init__.py first (drop the unconditional
Channel/Group/Message/Session table + MessagingService re-exports), then
deleted services/messaging.py, then trimmed db/seed.py to only create_agents
(create_channels/create_channel_memberships/create_initial_messages gone).

Forced expansion: api/routes/{channels,groups,sessions,messages}.py import
roboco.services.messaging directly (not through the package __init__), as
does api/routes/tasks.py (the session-links embed on GET /tasks/{id} and the
GET /{id}/sessions route). Deleting messaging.py without addressing these
breaks `import roboco.api.app` immediately, since app.py eagerly imports all
route modules at startup. Since the 4 CRUD route files are 100%
MessagingService-backed with zero independent logic (and are wholesale
deletes in the plan's later API-routes task anyway), deleted them now +
unmounted from app.py/routes/__init__.py; tasks.py got the same surgical
trim its later task already specified (drop session-links embed +
TaskSessionLinkResponse/TaskResponse.sessions). This pulls a slice of that
later work forward — the routes/schemas for channels/groups/sessions/messages
still need their own pass, but their messaging-coupled parts are gone.

Verified with a full-suite collection sweep (12010 tests collected, zero
import errors) beyond the directly touched test dirs, given the expanded
blast radius.

* refactor: remove channel/session/message models, tables, and channel policy

Models: deleted channel.py/group.py/session.py/messaging.py wholesale
(zero external consumers besides the models/__init__.py re-export).
message.py surgically trimmed: removed MessageCreate (dead) and MessageEdit
(never instantiated; ExtractedMessage.edit_history retyped to
list[dict[str, Any]] to match how it's actually persisted — confirmed
ExtractedMessage was never written to any DB table, so MessageTable's
removal carries no functional risk to the kept extraction pipeline).
base.py: removed SessionStatus + ChannelType, kept MessageType. Also
removed the confirmed-dead channels_read/channels_write fields from
models/agent.py:AgentPermissions and models/dashboard.py:ChannelFeedData.

db/tables.py: deleted ChannelTable/GroupTable/SessionTable/SessionTaskTable/
MessageTable, TaskTable.session_links, and JournalEntryTable.session_id —
cascaded through models/journal.py, services/journal.py, and
api/schemas+routes/journals.py (22 plumbing sites).

foundation/policy/communications.py: removed the ChannelSpec/CHANNELS
catalog + TEAM_SCOPED_ROLES/_CELL_*/_AUDITOR_ONLY helpers, kept the
notification policy (Priority/parse_priority/NOTIFY_SENDER_ROLES/
ACK_REQUIRED_BY_TYPE). enforcement/channel_access.py deleted (confirmed
fully dead in production). agents_config.py: removed CHANNEL_ACCESS
(kept A2A_ALLOWED_PAIRS). seeds/initial_data.py: removed
DEFAULT_CHANNELS/CHANNEL_MEMBERSHIPS/AUDITOR_SILENT_ACCESS + the
never-consumed INITIAL_MESSAGES. config.py: removed
session_idle_timeout_seconds (zero consumers). exceptions.py: removed
dead ChannelError/ChannelAccessDeniedError/SessionClosedError.

Forced expansion beyond the original file list — ChannelType cascaded
into a live, mounted surface the plan didn't trace: agents_config.
CHANNEL_ACCESS -> services/permissions.py's channel-RBAC methods (not
models/permissions.py, which turned out to have no channel code at all)
-> two real endpoints in api/routes/stream.py (GET /permissions,
GET /permissions/channel/{name}) and two dependency factories in
api/deps.py. Removed the channel methods + fields, deleted the
channel-specific stream.py endpoint, deleted require_channel_read/write.
Also deleted api/schemas/{channels,sessions}.py (hard dependency on the
removed enums; already fully dead after the Task 10 route deletions) and
api/schemas/messages.py (a TYPE_CHECKING-only import of the deleted
MessageTable; likewise already fully dead) + its dedicated test file.

Test updates: test_permissions.py -14 channel tests (matches the planned
count exactly), test_communications.py / test_communications_consumers.py
split to keep only notification-policy coverage, test_exceptions.py -9,
test_deps.py -4, plus the journal/stream/foundation-smoke fallout. Also
fixed a pre-existing (Task 7) broken assertion in
test_foundation_phase3_smoke.py that inspected a `say()` method already
removed from ContentActions.

Verified: full-suite collection (11961 tests, zero import errors) and a
complete test run (11567 passed, 394 skipped, 0 failed) in addition to
the targeted suites.

* migration: drop channels/groups/sessions/session_tasks/messages + enum types

alembic/versions/060_drop_messaging.py: drop_column journal_entries.
session_id (sidesteps hardcoding the FK constraint name — verified
empirically against a live migrated DB that it's actually
fk_journal_entries_session_id_sessions, but drop_column doesn't care
either way); drop_table in FK order (messages -> session_tasks ->
sessions -> groups -> channels); DROP TABLE IF EXISTS chunks_conversations
(runtime-provisioned, not alembic-managed, would otherwise orphan); DROP
TYPE IF EXISTS for messagetype/sessionstatus/sessionscope/channeltype
(messagetype's Python enum stays for ExtractedMessage, but the DB type
had zero live columns left once MessageTable was dropped in the prior
commit). downgrade() raises NotImplementedError — one-way removal.

Pruned scripts/reset_runtime_state.sql + .sh: removed the DELETE/COUNT
lines for messages/session_tasks/sessions/groups/channels and the
groups.active_session_id reset block.

Verified end-to-end against a scratch Postgres DB: full migration chain
001->060 applies cleanly, alembic heads shows a single head, all 6 dropped
tables + 4 enum types + the journal_entries.session_id column are
confirmed gone, journal_entries keeps only its journal_id/task_id FKs,
downgrade correctly raises NotImplementedError without corrupting DB
state, and the pruned reset_runtime_state.sql runs clean (no errors)
against a fully-migrated DB.

* refactor(api): remove channel/session/message routes + WS streams

Most of this task's file list was already forced through in earlier
commits (routes/{channels,groups,sessions,messages}.py + app.py/__init__.py
unmounting in the MessagingService-deletion commit; tasks.py's
session-links embed + GET /{id}/sessions + schemas/tasks.py's
TaskResponse.sessions in that same commit; deps.py's require_channel_read/
write + schemas/{channels,sessions}.py in the models/tables commit). This
closes out what was left:

- api/websocket.py: deleted the channel_stream + session_stream routes,
  ConnectionManager's channel_connections/session_connections dicts,
  connect_channel/connect_session, broadcast_to_channel/broadcast_to_session,
  get_channel_subscriber_count, and their cleanup lines in disconnect().
  Agent streams, notification streams, and the operator system stream are
  untouched.
- api/websocket_bridge.py: deleted _handle_session_event +
  _handle_message_event and their SESSION_CREATED/SESSION_CLOSED/
  SESSION_TIMEOUT/MESSAGE_SENT subscriptions. The A2A live-view, rate-limit,
  usage, agent-lifecycle, and notification bridges are untouched.
- api/schemas/websocket.py: removed NewMessageBroadcast, WSMessageNew,
  WSMessageEdit, WSMessageDelete, WSSessionClosed — kept the WSMessage base
  class (still subclassed by the kept WSAgentStream/WSNotification) plus
  those two.
- api/schemas/groups.py: deleted (already fully orphaned since routes/
  groups.py was removed; its GroupResponse/GroupDetailResponse had zero
  consumers).

Updated the 5 websocket test files accordingly (removed the channel/
session-specific tests + fixed imports); test_websocket_bridge.py's
registration-coverage test dropped the SESSION_*/MESSAGE_SENT assertions.

Verified: full-suite collection (11943 tests, zero import errors) and a
complete test run (11549 passed, 394 skipped, 0 failed).

* docs: retire channels/sessions/messages from agent-facing docs + CLAUDE.md

Rewrites docs/rag (RAG-indexed) + docs/map + CLAUDE.md to reflect A2A (dm +
read_a2a) as primary agent comms; deletes the channel docs, splits messaging-tools
+ messaging-notification (renamed notification.md), swaps the WS worked example to
A2A_MESSAGE_SENT. _complete_map.md still needs regeneration (generated file).

* refactor(panel): remove Communications surface (channels/sessions)

Deletes the /communications routes, message components, task-detail Sessions tab,
use-channels + channel/session WS hooks, and the channels/sessions/messages/groups
api clients; prunes the Channel/Session/Message/Group types + mock data. (Auditor
live-feeds + dashboard.ts dead-route cleanup is a follow-up.)

* refactor(panel): drop auditor channel-feed + dead communication-metric route

* docs(map): regenerate _complete_map from updated slices

* fix(a2a): reduce get_unread_messages complexity below xenon C + stale comments

Extract the per-conversation unread-counter recompute into _reset_unread_counter
(the CI quality gate flagged get_unread_messages as rank C). Also drop the deleted
open_session from a content_actions comment and reword an evidence_repo docstring
that cited the removed messaging._notify_mentions.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>
2026-07-04 03:10:33 +02:00

1334 lines
44 KiB
Python

"""TaskService coverage — final misc paths to push to 100%.
Targets:
- _append_capped truncation
- _default_claim_statuses for role-specific
- extract_original_developer invalid UUID
- _validate_parent_depth circular + max-depth + missing parent
- activate without project_id
- branch creation methods (with git mocks)
- claim role validations
- TaskLifecycleError catches in unclaim_for_reaper / unclaim_for_agent
- docs_complete branches (no notes / no assigned_to / has documenter)
- completion_notes early-return
- get_completing_agent_role missing agent
- complete main_pm escalation to CEO for root parent
- complete with PR not merged
- list_by_team_or_assignee with no conditions
- escalate_to_ceo_for_agent → inner escalate returns None
- mark_agent_idle missing agent
- qa_fail mismatch logging
- cell_pm_complete missing task
- get_task_service factory
"""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import AsyncMock, MagicMock
from uuid import UUID, uuid4
import pytest
import pytest_asyncio
from roboco.db.tables import AgentTable, ProjectTable, TaskTable, WorkSessionTable
from roboco.enforcement import TaskLifecycleError
from roboco.foundation.policy.content import markers
from roboco.models import AgentRole, AgentStatus, Team
from roboco.models.base import Complexity, TaskNature, TaskStatus, TaskType
from roboco.models.permissions import AgentContext
from roboco.models.task import TaskCreateRequest
from roboco.models.work_session import WorkSessionStatus
from roboco.services.base import ValidationError
from roboco.services.task import (
TaskService,
_append_capped,
_default_claim_statuses,
extract_original_developer,
get_task_service,
)
from roboco.templates.git.constants import MAX_TASK_DEPTH
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from sqlalchemy import Table
from sqlalchemy.ext.asyncio import AsyncSession
@pytest_asyncio.fixture
async def task_setup(
db_session: AsyncSession,
) -> AsyncIterator[dict]:
agent = AgentTable(
id=uuid4(),
name="Dev",
slug=f"be-dev-{uuid4().hex[:8]}",
role=AgentRole.DEVELOPER,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="dev",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(agent)
await db_session.flush()
project = ProjectTable(
id=uuid4(),
name="P",
slug=f"p-{uuid4().hex[:8]}",
git_url="https://example.com/r.git",
default_branch="main",
assigned_cell=Team.BACKEND,
created_by=agent.id,
)
db_session.add(project)
await db_session.flush()
yield {
"svc": TaskService(db_session),
"agent_id": agent.id,
"project_id": project.id,
"project_slug": project.slug,
"db": db_session,
}
def _req(setup: dict, **overrides: Any) -> TaskCreateRequest:
return TaskCreateRequest(
title=overrides.pop("title", "t"),
description=overrides.pop("description", "d"),
acceptance_criteria=overrides.pop("acceptance_criteria", ["ac"]),
team=overrides.pop("team", Team.BACKEND),
created_by=setup["agent_id"],
project_id=setup["project_id"],
task_type=overrides.pop("task_type", TaskType.CODE),
nature=overrides.pop("nature", TaskNature.TECHNICAL),
estimated_complexity=overrides.pop("estimated_complexity", Complexity.MEDIUM),
**overrides,
)
# ---------------------------------------------------------------------------
# Module-level helpers
# ---------------------------------------------------------------------------
def test_append_capped_truncates_when_over_max() -> None:
"""When the joined notes exceed _MAX_NOTES_CHARS, oldest gets dropped."""
big_existing = "OLD" * 4000 # ~12000 chars
addition = "newest"
result = _append_capped(big_existing, addition)
assert "[...earlier notes truncated for size...]" in result
assert result.endswith(addition)
def test_append_capped_no_truncation_below_max() -> None:
out = _append_capped("hello", "world")
assert out == "hello\n\nworld"
def test_default_claim_statuses_for_qa() -> None:
"""QA role returns the role-specific set."""
statuses = _default_claim_statuses("qa")
assert TaskStatus.AWAITING_QA in statuses
def test_default_claim_statuses_for_unknown_role() -> None:
statuses = _default_claim_statuses("developer")
assert TaskStatus.PENDING in statuses
assert TaskStatus.NEEDS_REVISION in statuses
def test_default_claim_statuses_for_none() -> None:
statuses = _default_claim_statuses(None)
assert statuses == {TaskStatus.PENDING}
def test_extract_original_developer_invalid_format() -> None:
"""Invalid UUID format returns None even when the marker is present."""
task = SimpleNamespace(orchestration_markers={"original_developer": "not-a-uuid"})
assert extract_original_developer(task) is None
def test_extract_original_developer_no_match() -> None:
task = SimpleNamespace(orchestration_markers={"documenter": "x"})
assert extract_original_developer(task) is None
def test_extract_original_developer_empty() -> None:
assert (
extract_original_developer(SimpleNamespace(orchestration_markers=None)) is None
)
assert extract_original_developer(SimpleNamespace(orchestration_markers={})) is None
def test_extract_original_developer_valid() -> None:
test_uuid = "12345678-1234-1234-1234-123456789012"
task = SimpleNamespace(orchestration_markers={"original_developer": test_uuid})
assert extract_original_developer(task) == test_uuid
# ---------------------------------------------------------------------------
# _validate_parent_depth
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_validate_parent_depth_missing_parent_raises(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
with pytest.raises(ValidationError, match="not found"):
await svc._validate_parent_depth(uuid4())
@pytest.mark.asyncio
async def test_validate_parent_depth_circular_reference(
task_setup: dict,
db_session: AsyncSession,
) -> None:
"""A self-referential parent loop raises ValidationError."""
svc = task_setup["svc"]
a = await svc.create(_req(task_setup))
b = await svc.create(_req(task_setup, parent_task_id=a.id))
# Force circular: a.parent_task_id = b.id
a.parent_task_id = b.id
await db_session.flush()
with pytest.raises(ValidationError, match="Circular reference"):
await svc._validate_parent_depth(a.id)
@pytest.mark.asyncio
async def test_validate_parent_depth_exceeds_max(
task_setup: dict,
) -> None:
"""Adding a child past MAX_TASK_DEPTH raises ValidationError."""
svc = task_setup["svc"]
# Build a chain of MAX_TASK_DEPTH+1 tasks
parent = None
for _ in range(MAX_TASK_DEPTH):
new = await svc.create(
_req(task_setup, parent_task_id=parent.id if parent else None)
)
parent = new
assert parent is not None
with pytest.raises(ValidationError, match="MAX_TASK_DEPTH"):
await svc._validate_parent_depth(parent.id)
# ---------------------------------------------------------------------------
# activate without project_id
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_activate_without_project_or_product_raises(
task_setup: dict, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A task with neither a project nor a product cannot be activated.
A task needs a project (a repo) OR a product (a cell->project map for a
fan-out/coordination task). Only when BOTH are missing is it genuinely
misconfigured and activation must raise. Stub get() to return a task with
project_id=None AND product_id=None so the guard is reached.
"""
svc = task_setup["svc"]
fake_task = MagicMock()
fake_task.id = uuid4()
fake_task.title = "fake"
fake_task.status = TaskStatus.BACKLOG
fake_task.project_id = None
fake_task.product_id = None
async def _stub_get(tid: Any) -> Any:
del tid
return fake_task
monkeypatch.setattr(svc, "get", _stub_get)
with pytest.raises(ValueError, match="no project or product"):
await svc.activate(fake_task.id, agent_role="cell_pm")
# ---------------------------------------------------------------------------
# _ensure_branch_for_task / _auto_create_branch / _resolve_parent_branch
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ensure_branch_returns_existing(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.branch_name = "feature/backend/EXISTING"
out = await svc._ensure_branch_for_task(task, task_setup["agent_id"])
assert out == "feature/backend/EXISTING"
@pytest.mark.asyncio
async def test_ensure_branch_no_project_id_raises(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
# Cleared project_id via raw SQL won't work here — instead, mock task.project_id
task.project_id = None
with pytest.raises(ValueError, match="project_id"):
await svc._ensure_branch_for_task(task, task_setup["agent_id"])
@pytest.mark.asyncio
async def test_ensure_branch_batch_umbrella_returns_empty(
task_setup: dict,
) -> None:
"""A MegaTask umbrella (batch_id, top-level, no project/product) is branchless
by design — it must short-circuit to "" rather than hit the misconfigured
raise, so the claim path treats it as coordination, not a defect."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.project_id = None
task.product_id = None
task.batch_id = uuid4()
task.parent_task_id = None
out = await svc._ensure_branch_for_task(task, task_setup["agent_id"])
assert out == ""
@pytest.mark.asyncio
async def test_create_denies_stray_batch_id_on_targeted_top_level_task(
task_setup: dict,
) -> None:
"""Guardrail: a top-level task that targets a project must NOT carry a
batch_id — that umbrella-shaped-but-targeted task would otherwise spoof the
branchless exemption. create() refuses it."""
svc = task_setup["svc"]
# _req sets project_id; adding batch_id with no parent is the spoof shape.
with pytest.raises(ValueError, match="batch_id is only valid"):
await svc.create(_req(task_setup, batch_id=uuid4()))
@pytest.mark.asyncio
async def test_create_denies_batch_root_subtask_under_non_umbrella_parent(
task_setup: dict,
) -> None:
"""Guardrail: a batch root-subtask's parent must be the batch umbrella. A
child pointed at a normal (non-umbrella) parent, or a mismatched batch, is
refused."""
svc = task_setup["svc"]
parent = await svc.create(_req(task_setup)) # a normal task, no batch_id
with pytest.raises(ValueError, match="parent must be the batch umbrella"):
await svc.create(_req(task_setup, batch_id=uuid4(), parent_task_id=parent.id))
@pytest.mark.asyncio
async def test_update_rejects_breaking_batch_umbrella_shape(
task_setup: dict,
) -> None:
"""Guardrail completeness: update() re-validates the MegaTask shape, so a
PATCH that adds a project to a branchless umbrella (which would spoof the
branch-gate / no-PR exemption) is refused — the invariant the branchless
predicates trust is enforced on mutation, not only at create."""
db = task_setup["db"]
svc = task_setup["svc"]
umbrella = TaskTable(
id=uuid4(),
title="MegaTask",
description="d",
acceptance_criteria=["ac"],
status=TaskStatus.PENDING,
priority=1,
task_type=TaskType.CODE,
nature=TaskNature.TECHNICAL,
project_id=None, # a valid umbrella targets neither
product_id=None,
batch_id=uuid4(),
parent_task_id=None,
team=Team.MAIN_PM,
created_by=task_setup["agent_id"],
)
db.add(umbrella)
await db.flush()
with pytest.raises(ValueError, match="MegaTask shape"):
await svc.update(umbrella.id, project_id=task_setup["project_id"])
@pytest.mark.asyncio
async def test_auto_create_branch_no_project_raises(
task_setup: dict,
monkeypatch: pytest.MonkeyPatch,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
fake_project_svc = MagicMock()
fake_project_svc.get = AsyncMock(return_value=None)
monkeypatch.setattr(
"roboco.services.project.get_project_service",
lambda _s: fake_project_svc,
)
monkeypatch.setattr(
"roboco.services.git.get_git_service",
lambda _s: MagicMock(),
)
with pytest.raises(ValueError, match="not found"):
await svc._auto_create_branch(task, task_setup["agent_id"])
@pytest.mark.asyncio
async def test_auto_create_branch_succeeds(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Full happy path of _auto_create_branch."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
await db_session.flush()
fake_project = MagicMock()
fake_project.id = task_setup["project_id"]
fake_project.slug = task_setup["project_slug"]
fake_project.default_branch = "main"
fake_project.assigned_cell = Team.BACKEND
fake_project_svc = MagicMock()
fake_project_svc.get = AsyncMock(return_value=fake_project)
fake_git = MagicMock()
fake_git.get_workspace = AsyncMock(return_value="/workspace")
fake_git.create_branch = AsyncMock(return_value=("feature/backend/AAAAAAA", None))
monkeypatch.setattr(
"roboco.services.project.get_project_service",
lambda _s: fake_project_svc,
)
monkeypatch.setattr("roboco.services.git.get_git_service", lambda _s: fake_git)
out = await svc._auto_create_branch(task, task_setup["agent_id"])
assert out == "feature/backend/AAAAAAA"
assert task.branch_name == "feature/backend/AAAAAAA"
@pytest.mark.asyncio
async def test_find_ancestor_branch_walks_up(
task_setup: dict, db_session: AsyncSession
) -> None:
svc = task_setup["svc"]
grand = await svc.create(_req(task_setup))
grand.branch_name = "feature/backend/GRAND"
parent = await svc.create(_req(task_setup, parent_task_id=grand.id))
# Parent has no branch_name
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
await db_session.flush()
out = await svc._find_ancestor_branch(child)
assert out == "feature/backend/GRAND"
@pytest.mark.asyncio
async def test_find_ancestor_branch_handles_circular(
task_setup: dict, db_session: AsyncSession
) -> None:
"""Circular reference returns None gracefully (just logs warning)."""
svc = task_setup["svc"]
a = await svc.create(_req(task_setup))
b = await svc.create(_req(task_setup, parent_task_id=a.id))
a.parent_task_id = b.id
await db_session.flush()
out = await svc._find_ancestor_branch(b)
assert out is None
@pytest.mark.asyncio
async def test_find_ancestor_branch_returns_none_when_no_branches(
task_setup: dict, db_session: AsyncSession
) -> None:
"""When parent chain has no branches anywhere, returns None."""
svc = task_setup["svc"]
parent = await svc.create(_req(task_setup))
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
await db_session.flush()
out = await svc._find_ancestor_branch(child)
assert out is None
@pytest.mark.asyncio
async def test_resolve_parent_branch_uses_default_when_no_ancestor(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
fake_project = MagicMock()
fake_project.default_branch = "main"
out = await svc._resolve_parent_branch(task, fake_project)
assert out == "main"
@pytest.mark.asyncio
async def test_resolve_parent_branch_uses_ancestor(
task_setup: dict, db_session: AsyncSession
) -> None:
svc = task_setup["svc"]
parent = await svc.create(_req(task_setup))
parent.branch_name = "feature/backend/PARENT"
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
await db_session.flush()
fake_project = MagicMock()
fake_project.default_branch = "main"
out = await svc._resolve_parent_branch(child, fake_project)
assert out == "feature/backend/PARENT"
# ---------------------------------------------------------------------------
# _resolve_team_dir branches
# ---------------------------------------------------------------------------
def test_resolve_team_dir_fullstack(task_setup: dict) -> None:
svc = task_setup["svc"]
fake_project = MagicMock()
fake_project.assigned_cell = Team.FULLSTACK
fake_project.slug = "p"
task = MagicMock()
task.team = Team.BACKEND
out = svc._resolve_team_dir(fake_project, task)
assert "p/backend" in out
def test_resolve_team_dir_fullstack_no_team(task_setup: dict) -> None:
svc = task_setup["svc"]
fake_project = MagicMock()
fake_project.assigned_cell = Team.FULLSTACK
fake_project.slug = "p"
task = MagicMock()
task.team = None
out = svc._resolve_team_dir(fake_project, task)
assert "p/cross" in out
def test_resolve_team_dir_task_team(task_setup: dict) -> None:
svc = task_setup["svc"]
fake_project = MagicMock()
fake_project.assigned_cell = Team.BACKEND
task = MagicMock()
task.team = Team.FRONTEND
out = svc._resolve_team_dir(fake_project, task)
assert out == "frontend"
def test_resolve_team_dir_falls_back_to_project_cell(task_setup: dict) -> None:
svc = task_setup["svc"]
fake_project = MagicMock()
fake_project.assigned_cell = Team.BACKEND
task = MagicMock()
task.team = None
out = svc._resolve_team_dir(fake_project, task)
assert out == "backend"
def test_resolve_team_dir_cross_when_nothing(task_setup: dict) -> None:
svc = task_setup["svc"]
fake_project = MagicMock()
fake_project.assigned_cell = None
task = MagicMock()
task.team = None
out = svc._resolve_team_dir(fake_project, task)
assert out == "cross"
# ---------------------------------------------------------------------------
# _validate_claim_status — role-specific status missing agent
# ---------------------------------------------------------------------------
def test_validate_claim_status_no_agent_role_specific(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.status = TaskStatus.AWAITING_QA
err = svc._validate_claim_status(task, agent=None, valid_statuses=set())
assert err is not None
assert "role required" in err
def test_validate_claim_status_invalid(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.status = TaskStatus.COMPLETED
err = svc._validate_claim_status(
task, agent=None, valid_statuses={TaskStatus.PENDING}
)
assert err == "invalid status for role"
def test_validate_claim_status_ok(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.status = TaskStatus.PENDING
err = svc._validate_claim_status(
task, agent=None, valid_statuses={TaskStatus.PENDING}
)
assert err is None
def test_validate_not_self_review_no_agent(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
out = svc._validate_not_self_review(task, agent=None, agent_id=uuid4())
assert out is None
def test_validate_not_self_review_no_role(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
agent = MagicMock(role=None)
out = svc._validate_not_self_review(task, agent, agent_id=uuid4())
assert out is None
def test_validate_not_self_review_dev_role(task_setup: dict) -> None:
"""Dev role isn't QA/documenter so passes (returns None)."""
svc = task_setup["svc"]
task = MagicMock()
agent = MagicMock(role=AgentRole.DEVELOPER)
out = svc._validate_not_self_review(task, agent, agent_id=uuid4())
assert out is None
def test_validate_not_self_review_qa_self(task_setup: dict) -> None:
svc = task_setup["svc"]
aid = uuid4()
task = MagicMock()
task.orchestration_markers = {"original_developer": str(aid)}
agent = MagicMock(role=AgentRole.QA)
out = svc._validate_not_self_review(task, agent, agent_id=aid)
assert "self-review" in (out or "")
def test_set_original_developer_skips_when_already_set(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.orchestration_markers = {"original_developer": "already-set"}
task.assigned_to = uuid4()
agent = MagicMock(role=AgentRole.QA, id=uuid4())
# An existing original_developer marker must not be overwritten.
before = dict(task.orchestration_markers)
svc._set_original_developer_context(task, agent)
assert task.orchestration_markers == before
def test_set_original_developer_skips_when_no_role(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
agent = MagicMock(role=None)
svc._set_original_developer_context(task, agent)
def test_set_original_developer_skips_when_dev_role(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = ""
task.assigned_to = uuid4()
agent = MagicMock(role=AgentRole.DEVELOPER, id=uuid4())
svc._set_original_developer_context(task, agent)
# dev role → early return, quick_context unchanged
assert task.quick_context == ""
def test_set_original_developer_skips_when_self(task_setup: dict) -> None:
svc = task_setup["svc"]
aid = uuid4()
task = MagicMock()
task.quick_context = ""
task.assigned_to = aid
agent = MagicMock(role=AgentRole.QA, id=aid)
svc._set_original_developer_context(task, agent)
# Same agent → don't set
assert task.quick_context == ""
# ---------------------------------------------------------------------------
# unclaim_for_reaper / unclaim_for_agent — TaskLifecycleError catches
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unclaim_for_reaper_lifecycle_error_returns(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Force the lifecycle validator to raise TaskLifecycleError → return."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.CLAIMED
task.assigned_to = task_setup["agent_id"]
await db_session.flush()
def _raise(*_args: Any, **_kwargs: Any) -> None:
raise TaskLifecycleError(
current_status="claimed",
target_status="pending",
valid_transitions=[],
)
monkeypatch.setattr(svc, "_validate_and_set_status", _raise)
# Should not raise — silently returns
await svc.unclaim_for_reaper(task.id)
@pytest.mark.asyncio
async def test_unclaim_for_agent_lifecycle_error_returns_none(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.CLAIMED
task.assigned_to = task_setup["agent_id"]
await db_session.flush()
def _raise(*_args: Any, **_kwargs: Any) -> None:
raise TaskLifecycleError(
current_status="claimed",
target_status="pending",
valid_transitions=[],
)
monkeypatch.setattr(svc, "_validate_and_set_status", _raise)
out = await svc.unclaim_for_agent(task.id, agent_id=task_setup["agent_id"])
assert out is None
@pytest.mark.asyncio
async def test_resume_for_agent_lifecycle_error_returns_none(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.PAUSED
task.assigned_to = task_setup["agent_id"]
await db_session.flush()
async def _bad_resume(*_a: Any, **_kw: Any) -> None:
raise TaskLifecycleError(
current_status="paused",
target_status="in_progress",
valid_transitions=[],
)
monkeypatch.setattr(svc, "resume", _bad_resume)
out = await svc.resume_for_agent(task.id, agent_id=task_setup["agent_id"])
assert out is None
# ---------------------------------------------------------------------------
# docs_complete edge cases
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_docs_complete_indexes_when_documents_present(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Path: docs_complete spawns _index_docs_background when docs present."""
svc = task_setup["svc"]
doc = AgentTable(
id=uuid4(),
name="Doc",
slug=f"be-doc-{uuid4().hex[:8]}",
role=AgentRole.DOCUMENTER,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="d",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(doc)
await db_session.flush()
task = await svc.create(_req(task_setup))
task.status = TaskStatus.AWAITING_DOCUMENTATION
task.assigned_to = doc.id
task.pr_number = 1
task.pr_url = "u"
task.documents = [{"path": "doc1.md"}]
await db_session.flush()
fake_optimal = MagicMock()
fake_optimal.index_documentation = AsyncMock(return_value=1)
async def _get_optimal() -> Any:
return fake_optimal
monkeypatch.setattr("roboco.services.optimal.get_optimal_service", _get_optimal)
out = await svc.docs_complete(task.id, doc_notes="docs done")
assert out is not None
await asyncio.sleep(0.05)
def test_record_doc_notes_skips_empty(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = "before"
svc._record_doc_notes(task, None)
assert task.quick_context == "before"
svc._record_doc_notes(task, "")
assert task.quick_context == "before"
def test_record_documenter_context_skips_no_assignee(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.assigned_to = None
task.quick_context = "x"
svc._record_documenter_context(task)
assert task.quick_context == "x"
def test_record_documenter_context_skips_when_already(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.assigned_to = uuid4()
task.orchestration_markers = {"documenter": "something"}
before = dict(task.orchestration_markers)
svc._record_documenter_context(task)
assert task.orchestration_markers == before
def test_record_documenter_context_appends(task_setup: dict) -> None:
svc = task_setup["svc"]
aid = uuid4()
task = MagicMock()
task.assigned_to = aid
task.quick_context = "existing"
task.orchestration_markers = None
svc._record_documenter_context(task)
assert markers.get_documenter(task) == str(aid)
assert task.quick_context == "existing" # human field untouched
def test_record_documenter_context_first_entry(task_setup: dict) -> None:
svc = task_setup["svc"]
aid = uuid4()
task = MagicMock()
task.assigned_to = aid
task.orchestration_markers = None
svc._record_documenter_context(task)
assert markers.get_documenter(task) == str(aid)
def test_record_completion_notes_skips_empty(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = "before"
svc._record_completion_notes(task, None)
assert task.quick_context == "before"
def test_record_completion_notes_with_existing_context(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = "existing"
task.orchestration_markers = None
svc._record_completion_notes(task, "merged successfully")
# quick_context (the ResumptionNote slot) is left untouched; the note goes
# to a structured marker, not packed in as `completion_notes:<text>` soup.
assert task.quick_context == "existing"
assert markers.get_transition_note(task, "completion") == "merged successfully"
def test_record_completion_notes_no_existing_context(task_setup: dict) -> None:
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = None
task.orchestration_markers = None
svc._record_completion_notes(task, "merged")
assert task.quick_context is None
assert markers.get_transition_note(task, "completion") == "merged"
# ---------------------------------------------------------------------------
# _resolve_pm_for_review — direct candidate found
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_resolve_pm_for_review_finds_first_assignee(
task_setup: dict, db_session: AsyncSession
) -> None:
svc = task_setup["svc"]
pm_agent = AgentTable(
id=uuid4(),
name="PM",
slug=f"be-pm-{uuid4().hex[:8]}",
role=AgentRole.CELL_PM,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(pm_agent)
await db_session.flush()
parent = await svc.create(_req(task_setup, assigned_to=pm_agent.id))
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
pm_id = await svc._resolve_pm_for_review(child)
assert pm_id == pm_agent.id
# ---------------------------------------------------------------------------
# _get_completing_agent_role
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_completing_agent_role_none_for_missing_agent_id(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
out = await svc._get_completing_agent_role(None)
assert out is None
@pytest.mark.asyncio
async def test_get_completing_agent_role_none_for_unknown(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
out = await svc._get_completing_agent_role(uuid4())
assert out is None
@pytest.mark.asyncio
async def test_get_completing_agent_role_returns_role(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
out = await svc._get_completing_agent_role(task_setup["agent_id"])
assert out == "developer"
# ---------------------------------------------------------------------------
# complete: main_pm root parent escalates to CEO
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_complete_main_pm_root_parent_escalates_ceo(
task_setup: dict, db_session: AsyncSession
) -> None:
svc = task_setup["svc"]
main_pm = AgentTable(
id=uuid4(),
name="MainPM",
slug=f"main-pm-{uuid4().hex[:8]}",
role=AgentRole.MAIN_PM,
team=Team.MAIN_PM,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(main_pm)
await db_session.flush()
parent = await svc.create(_req(task_setup))
parent.status = TaskStatus.AWAITING_PM_REVIEW
parent.pr_number = 1
parent.pr_url = "u"
parent.pr_created = True
parent.docs_complete = True
await db_session.flush()
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
child.status = TaskStatus.COMPLETED
await db_session.flush()
out = await svc.complete(parent.id, agent_id=main_pm.id)
assert out is not None
# Root parent + main_pm + descendants → escalate to CEO
assert out.status == TaskStatus.AWAITING_CEO_APPROVAL
# ---------------------------------------------------------------------------
# complete: PR not merged returns None
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_complete_returns_none_when_pr_not_merged(
task_setup: dict, db_session: AsyncSession
) -> None:
"""Cell PM completing → escalate-to-Main PM is short-circuited so the
PR-not-merged guard is reached. Strip leaked Main PMs from earlier tests
that committed and survived rollback isolation.
"""
svc = task_setup["svc"]
await db_session.execute(
cast("Table", AgentTable.__table__)
.update()
.where(AgentTable.role == AgentRole.MAIN_PM)
.values(role=AgentRole.SYSTEM)
)
pm = AgentTable(
id=uuid4(),
name="PM",
slug=f"be-pm-{uuid4().hex[:8]}",
role=AgentRole.CELL_PM,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="pm",
capabilities=[],
permissions={},
metrics={},
)
db_session.add(pm)
await db_session.flush()
task = await svc.create(_req(task_setup))
task.status = TaskStatus.AWAITING_PM_REVIEW
ws = WorkSessionTable(
id=uuid4(),
project_id=task_setup["project_id"],
task_id=task.id,
agent_id=task_setup["agent_id"],
branch_name="feature/backend/x",
base_branch="main",
target_branch="main",
status=WorkSessionStatus.ACTIVE,
pr_status="open", # NOT merged
)
db_session.add(ws)
await db_session.flush()
task.work_session_id = ws.id
await db_session.flush()
out = await svc.complete(task.id, agent_id=pm.id)
assert out is None
# ---------------------------------------------------------------------------
# list_by_team_or_assignee — no conditions returns []
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_by_team_or_assignee_no_conditions_empty(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
rows = await svc.list_by_team_or_assignee(team=None, agent_id=None)
assert rows == []
@pytest.mark.asyncio
async def test_list_by_team_or_assignee_with_status(task_setup: dict) -> None:
svc = task_setup["svc"]
rows = await svc.list_by_team_or_assignee(
team=Team.BACKEND, agent_id=None, status=TaskStatus.PENDING
)
assert isinstance(rows, list)
# ---------------------------------------------------------------------------
# escalate_to_ceo_for_agent — inner escalate returns None → ValidationError
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_escalate_to_ceo_for_agent_inner_returns_none(
task_setup: dict,
db_session: AsyncSession,
) -> None:
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.AWAITING_PM_REVIEW
task.pr_number = 1
task.pr_url = "u"
task.pr_created = True
task.docs_complete = True
await db_session.flush()
agent_ctx = AgentContext(
agent_id=task_setup["agent_id"],
role=AgentRole.MAIN_PM,
team=Team.MAIN_PM,
slug="x",
)
class _P:
def can_perform_task_action(self, *a: Any, **kw: Any) -> bool:
del a, kw
return True
# Force the inner escalate_to_ceo to return None
object.__setattr__(svc, "escalate_to_ceo", AsyncMock(return_value=None))
with pytest.raises(ValidationError, match="awaiting_pm_review"):
await svc.escalate_to_ceo_for_agent(
task.id,
agent_ctx,
_P(),
"Substantial reasons for CEO review of breaking changes",
)
# ---------------------------------------------------------------------------
# mark_agent_idle — missing agent
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_mark_agent_idle_missing_agent_no_op(task_setup: dict) -> None:
svc = task_setup["svc"]
# Should not raise
await svc.mark_agent_idle(uuid4())
# ---------------------------------------------------------------------------
# qa_fail — actor mismatch logs warning
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_qa_fail_actor_mismatch_warning(
task_setup: dict, db_session: AsyncSession
) -> None:
svc = task_setup["svc"]
qa_a = AgentTable(
id=uuid4(),
name="QA-A",
slug=f"qa-a-{uuid4().hex[:8]}",
role=AgentRole.QA,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="qa",
capabilities=[],
permissions={},
metrics={},
)
qa_b = AgentTable(
id=uuid4(),
name="QA-B",
slug=f"qa-b-{uuid4().hex[:8]}",
role=AgentRole.QA,
team=Team.BACKEND,
status=AgentStatus.ACTIVE,
model_config={},
system_prompt="qa",
capabilities=[],
permissions={},
metrics={},
)
db_session.add_all([qa_a, qa_b])
await db_session.flush()
task = await svc.create(_req(task_setup))
task.status = TaskStatus.AWAITING_QA
task.claimed_by = qa_a.id # Different from qa_b
await db_session.flush()
out = await svc.qa_fail(qa_b.id, task.id, "needs work", issues=["x"])
assert out is not None # warning logged but flow continues
# ---------------------------------------------------------------------------
# cell_pm_complete — missing task returns None
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_cell_pm_complete_missing_task_returns_none(
task_setup: dict,
) -> None:
svc = task_setup["svc"]
out = await svc.cell_pm_complete(
task_setup["agent_id"], uuid4(), "merged", merge_commit="x"
)
assert out is None
# ---------------------------------------------------------------------------
# get_task_service factory
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_task_service_returns_instance(
db_session: AsyncSession,
) -> None:
out = get_task_service(db_session)
assert isinstance(out, TaskService)
# ---------------------------------------------------------------------------
# Final coverage gaps (line-specific)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ensure_branch_calls_auto_create(
task_setup: dict, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Cover line 595: _ensure_branch_for_task → _auto_create_branch."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
# No branch_name set → falls through to auto-create
async def _stub(_t: Any, _a: Any) -> str:
return "feature/backend/MOCK"
monkeypatch.setattr(svc, "_auto_create_branch", _stub)
out = await svc._ensure_branch_for_task(task, task_setup["agent_id"])
assert out == "feature/backend/MOCK"
@pytest.mark.asyncio
async def test_find_ancestor_branch_break_when_parent_missing(
task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Cover line 622: break when parent lookup returns None."""
svc = task_setup["svc"]
parent = await svc.create(_req(task_setup))
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
await db_session.flush()
real_get = svc.get
async def _stub_get(tid: Any) -> Any:
# Return None when looking up the parent
if tid == parent.id:
return None
return await real_get(tid)
monkeypatch.setattr(svc, "get", _stub_get)
out = await svc._find_ancestor_branch(child)
assert out is None
def test_validate_claim_team_no_agent(task_setup: dict) -> None:
"""Cover line 838: _validate_claim_team early return when no agent."""
svc = task_setup["svc"]
task = MagicMock()
task.team = Team.BACKEND
out = svc._validate_claim_team(task, agent=None)
assert out is None
def test_validate_not_self_review_qa_with_different_dev(task_setup: dict) -> None:
"""Cover line 866: _validate_not_self_review returns None when QA differs."""
svc = task_setup["svc"]
task = MagicMock()
task.quick_context = f"original_developer:{uuid4()}"
agent = MagicMock(role=AgentRole.QA)
out = svc._validate_not_self_review(task, agent, agent_id=uuid4())
assert out is None
def test_set_original_developer_records_when_different(task_setup: dict) -> None:
"""Sets the original_developer marker when assigned_to != agent.id."""
svc = task_setup["svc"]
task = MagicMock()
task.orchestration_markers = None
other_id = uuid4()
task.assigned_to = other_id
agent = MagicMock(role=AgentRole.QA, id=uuid4())
svc._set_original_developer_context(task, agent)
assert markers.get_original_developer(task) == str(other_id)
@pytest.mark.asyncio
async def test_finalize_claim_refreshes_after_branch_creation(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Cover line 998: session.refresh after _ensure_branch_for_task succeeds."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
await db_session.flush()
async def _ensure_branch(_t: Any, _a: Any) -> str:
_t.branch_name = "feature/backend/X"
return "feature/backend/X"
monkeypatch.setattr(svc, "_ensure_branch_for_task", _ensure_branch)
refresh_mock = AsyncMock()
monkeypatch.setattr(svc.session, "refresh", refresh_mock)
claimed = await svc.claim(task.id, task_setup["agent_id"])
assert claimed is not None
refresh_mock.assert_awaited()
@pytest.mark.asyncio
async def test_resolve_pm_for_review_returns_none_when_chain_broken(
task_setup: dict,
db_session: AsyncSession,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Cover line 2650: parent lookup returns None mid-chain."""
svc = task_setup["svc"]
parent = await svc.create(_req(task_setup))
child = await svc.create(_req(task_setup, parent_task_id=parent.id))
await db_session.flush()
real_get = svc.get
async def _stub_get(tid: Any) -> Any:
# Return None when looking up the parent
if tid == parent.id:
return None
return await real_get(tid)
monkeypatch.setattr(svc, "get", _stub_get)
out = await svc._resolve_pm_for_review(child)
assert out is None
@pytest.mark.asyncio
async def test_unclaim_for_agent_works_with_uuid_round_trip(
task_setup: dict,
db_session: AsyncSession,
) -> None:
"""Regression for 2026-05-08: main-pm got 'not your claim' on its OWN
claim. Pin the UUID-comparator behavior: same agent_id in (whether
fresh UUID or string-coerced UUID round-trip), unclaim succeeds.
"""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.CLAIMED
task.assigned_to = task_setup["agent_id"]
await db_session.flush()
# Round-trip the UUID through string → UUID to mirror what an HTTP
# request body / header would do via Pydantic. The comparator must
# treat both the freshly-constructed and the round-tripped UUID as
# equal (which Python's UUID class does — pinning so a future move
# to a custom comparator can't silently break it).
same_uuid_via_str = UUID(str(task_setup["agent_id"]))
out = await svc.unclaim_for_agent(task.id, agent_id=same_uuid_via_str)
assert out is not None, "round-tripped UUID rejected — comparator broken"
assert out.assigned_to is None
assert out.status == TaskStatus.PENDING
@pytest.mark.asyncio
async def test_resume_for_agent_works_with_uuid_round_trip(
task_setup: dict,
db_session: AsyncSession,
) -> None:
"""Mirror of the unclaim regression for resume_for_agent."""
svc = task_setup["svc"]
task = await svc.create(_req(task_setup))
task.status = TaskStatus.PAUSED
task.assigned_to = task_setup["agent_id"]
await db_session.flush()
same_uuid_via_str = UUID(str(task_setup["agent_id"]))
out = await svc.resume_for_agent(task.id, agent_id=same_uuid_via_str)
assert out is not None, "round-tripped UUID rejected — comparator broken"
assert out.status == TaskStatus.IN_PROGRESS