Fix: CEO reject conduit and main pm routing (#91)

* fix(tasks): deliver CEO change-requests to the reworker and route integration rejects to the Main PM

ceo_reject previously appended the CEO's required-changes only to quick_context
(never surfaced to an agent) and always reassigned to the original developer, so
the feedback never reached whoever reworked the task and a coordination/integration
root went to a developer instead of the Main PM.

Now ceo_reject records the reason as a DECISION_LOG handoff journal entry — the
channel evidence/journal_highlights already serves to the task's assignee — and
routes a coordination task (no project, has product) to team=main_pm + the Main PM
so it can delegate the rework. Leaf dev tasks still return to the original
developer. The journal write is best-effort (author-exists guard) and never blocks
a reject. Adds tests for the routing and the handoff-journal write.

* feat(gateway): wire the agent context_briefing receive-path

EvidenceRepo's agent-scoped methods were stubbed to return empty lists, so agents
never received notifications, A2A DMs, @mentions, recent team activity, or in-lane
blockers through their briefing — the system was effectively send-only. This
implements all five as single capped queries over the live tables (plus a light
task-metadata-gap check), mirroring the existing journal_highlights query. Each runs
on the per-verb briefing path, so each stays a single LIMIT-10 indexed lookup.

Unit tests cover the row mapping + empty paths; an integration suite exercises the
real SQL (array contains, the a2a slug filter, team+status) against Postgres.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

roboco/services/gateway/evidence_repo.py

@@ -1,6 +1,8 @@
-"""EvidenceRepo: aggregates unread A2As, mentions, notifications, etc.
+"""EvidenceRepo: aggregates unread A2As, mentions, notifications, and task/team
+context for an agent's ``context_briefing`` (plus the task journal handoff).
 
-Phase 1 stub returns empty lists; Phase 2+ wires real queries.
+Each method is a single capped query over the live tables — they run on the
+briefing-assembly path (per verb), so each stays cheap (LIMIT 10, indexed lookups).
 """
 
 from __future__ import annotations
@@ -18,28 +20,216 @@ class EvidenceRepo:
         self._db = db_session
 
     async def list_unread_a2a(self, agent_id: UUID) -> list[dict[str, Any]]:
-        del agent_id
+        """Open A2A conversations with unread messages for this agent.
-        return []
+
+        Conversations are keyed by agent slug (``agent_a``/``agent_b``) with a
+        per-side unread counter; surface the ones this agent has yet to read.
+        """
+        from sqlalchemy import or_, select
+
+        from roboco.db.tables import A2AConversationTable, AgentTable
+
+        slug = await self._db.scalar(
+            select(AgentTable.slug).where(AgentTable.id == agent_id)
+        )
+        if slug is None:
+            return []
+        rows = (
+            (
+                await self._db.execute(
+                    select(A2AConversationTable)
+                    .where(
+                        or_(
+                            (A2AConversationTable.agent_a == slug)
+                            & (A2AConversationTable.unread_by_a > 0),
+                            (A2AConversationTable.agent_b == slug)
+                            & (A2AConversationTable.unread_by_b > 0),
+                        )
+                    )
+                    .order_by(A2AConversationTable.updated_at.desc())
+                    .limit(10)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        items: list[dict[str, Any]] = []
+        for c in rows:
+            is_a = c.agent_a == slug
+            items.append(
+                {
+                    "conversation_id": str(c.id),
+                    "from_agent": c.agent_b if is_a else c.agent_a,
+                    "unread": c.unread_by_a if is_a else c.unread_by_b,
+                    "topic": c.topic,
+                    "task_id": str(c.task_id) if c.task_id else None,
+                }
+            )
+        return items
 
     async def list_unread_mentions(self, agent_id: UUID) -> list[dict[str, Any]]:
-        del agent_id
+        """Recent channel messages that @mention this agent.
-        return []
+
+        Messages carry no per-recipient read state, so this surfaces the most
+        recent mentions (capped); the briefing is rebuilt each turn.
+        """
+        from sqlalchemy import select
+
+        from roboco.db.tables import MessageTable
+
+        result = await self._db.execute(
+            select(
+                MessageTable.id,
+                MessageTable.agent_id,
+                MessageTable.channel_id,
+                MessageTable.content,
+                MessageTable.task_id,
+                MessageTable.timestamp,
+            )
+            .where(MessageTable.mentions.contains([agent_id]))
+            .order_by(MessageTable.timestamp.desc())
+            .limit(10)
+        )
+        return [
+            {
+                "message_id": str(row.id),
+                "from_agent": str(row.agent_id),
+                "channel_id": str(row.channel_id) if row.channel_id else None,
+                "excerpt": (row.content or "")[:280],
+                "task_id": str(row.task_id) if row.task_id else None,
+                "timestamp": row.timestamp.isoformat() if row.timestamp else None,
+            }
+            for row in result.all()
+        ]
 
     async def list_pending_notifications(self, agent_id: UUID) -> list[dict[str, Any]]:
-        del agent_id
+        """Unacknowledged, unexpired notifications addressed to this agent."""
-        return []
+        from datetime import UTC, datetime
+
+        from sqlalchemy import or_, select
+
+        from roboco.db.tables import NotificationTable
+
+        now = datetime.now(UTC)
+        result = await self._db.execute(
+            select(
+                NotificationTable.id,
+                NotificationTable.type,
+                NotificationTable.priority,
+                NotificationTable.subject,
+                NotificationTable.body,
+                NotificationTable.from_agent,
+                NotificationTable.related_task_id,
+                NotificationTable.timestamp,
+            )
+            .where(NotificationTable.to_agents.contains([agent_id]))
+            .where(~NotificationTable.acked_by.contains([agent_id]))
+            .where(
+                or_(
+                    NotificationTable.expires_at.is_(None),
+                    NotificationTable.expires_at > now,
+                )
+            )
+            .order_by(NotificationTable.timestamp.desc())
+            .limit(10)
+        )
+        return [
+            {
+                "notification_id": str(row.id),
+                "type": str(row.type),
+                "priority": str(row.priority),
+                "subject": row.subject,
+                "body": row.body,
+                "from_agent": str(row.from_agent) if row.from_agent else None,
+                "task_id": str(row.related_task_id) if row.related_task_id else None,
+                "timestamp": row.timestamp.isoformat() if row.timestamp else None,
+            }
+            for row in result.all()
+        ]
 
     async def task_metadata_gaps(self, task_id: UUID) -> list[str]:
-        del task_id
+        """Human-readable gaps in a task's metadata the owner should fill."""
-        return []
+        from sqlalchemy import select
+
+        from roboco.db.tables import TaskTable
+
+        task = await self._db.scalar(select(TaskTable).where(TaskTable.id == task_id))
+        if task is None:
+            return []
+        gaps: list[str] = []
+        if not task.acceptance_criteria:
+            gaps.append("no acceptance criteria")
+        if not task.description:
+            gaps.append("no description")
+        return gaps
 
     async def recent_team_activity(self, agent_id: UUID) -> list[dict[str, Any]]:
-        del agent_id
+        """Recently-updated tasks in this agent's team (lane awareness)."""
-        return []
+        from sqlalchemy import func, select
+
+        from roboco.db.tables import AgentTable, TaskTable
+
+        team = await self._db.scalar(
+            select(AgentTable.team).where(AgentTable.id == agent_id)
+        )
+        if team is None:
+            return []
+        result = await self._db.execute(
+            select(
+                TaskTable.id,
+                TaskTable.title,
+                TaskTable.status,
+                TaskTable.assigned_to,
+                TaskTable.updated_at,
+            )
+            .where(TaskTable.team == team)
+            .order_by(func.coalesce(TaskTable.updated_at, TaskTable.created_at).desc())
+            .limit(10)
+        )
+        return [
+            {
+                "task_id": str(row.id),
+                "title": row.title,
+                "status": str(row.status),
+                "assigned_to": str(row.assigned_to) if row.assigned_to else None,
+                "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+            }
+            for row in result.all()
+        ]
 
     async def blockers_in_lane(self, agent_id: UUID) -> list[dict[str, Any]]:
-        del agent_id
+        """Blocked tasks in this agent's team."""
-        return []
+        from sqlalchemy import select
+
+        from roboco.db.tables import AgentTable, TaskTable
+        from roboco.models.base import TaskStatus
+
+        team = await self._db.scalar(
+            select(AgentTable.team).where(AgentTable.id == agent_id)
+        )
+        if team is None:
+            return []
+        result = await self._db.execute(
+            select(
+                TaskTable.id,
+                TaskTable.title,
+                TaskTable.assigned_to,
+                TaskTable.dependency_ids,
+            )
+            .where(TaskTable.team == team)
+            .where(TaskTable.status == TaskStatus.BLOCKED)
+            .order_by(TaskTable.updated_at.desc())
+            .limit(10)
+        )
+        return [
+            {
+                "task_id": str(row.id),
+                "title": row.title,
+                "assigned_to": str(row.assigned_to) if row.assigned_to else None,
+                "blocked_on": [str(d) for d in (row.dependency_ids or [])],
+            }
+            for row in result.all()
+        ]
 
     async def journal_highlights_for_task(self, task_id: UUID) -> list[dict[str, Any]]:
         """The task's upstream handoff: every author's decision / reflection /

roboco/services/task.py

@@ -16,6 +16,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from roboco.db.tables import (
     AgentTable,
+    JournalEntryTable,
+    JournalTable,
     ProjectTable,
     SessionTaskTable,
     TaskTable,
@@ -34,6 +36,7 @@ from roboco.models.base import (
     AgentRole,
     AgentStatus,
     BlockerResolverType,
+    JournalEntryType,
     TaskNature,
     TaskStatus,
     TaskType,
@@ -42,6 +45,7 @@ from roboco.models.base import (
 from roboco.models.permissions import AgentContext, TaskAction
 from roboco.models.task import TaskCreateRequest
 from roboco.models.work_session import WorkSessionStatus
+from roboco.seeds.initial_data import AGENT_UUIDS
 from roboco.services.base import (
     BaseService,
     NotFoundError,
@@ -3819,6 +3823,48 @@ class TaskService(BaseService):
         await self.session.flush()
         return True
 
+    async def _write_handoff_journal(
+        self, *, author_id: UUID, task_id: UUID, title: str, content: str
+    ) -> None:
+        """Record a handoff journal entry from ``author_id`` for ``task_id``.
+
+        Handoff-typed (``DECISION_LOG``) entries are the channel that actually
+        reaches a downstream worker: ``EvidenceRepo.journal_highlights_for_task``
+        serves them into the assignee's evidence/briefing. ``quick_context`` is
+        write-only by comparison. Adds rows to the session WITHOUT committing — the
+        caller owns the transaction boundary.
+        """
+        # Best-effort: never let a journaling hiccup block the reject. In
+        # production the author (CEO) is a seeded agent; guard so a missing author
+        # (e.g. a minimal test DB) skips cleanly instead of FK-failing the flush.
+        author_exists = await self.session.scalar(
+            select(AgentTable.id).where(AgentTable.id == author_id)
+        )
+        if author_exists is None:
+            self.log.warning(
+                "Handoff journal skipped - author agent not found",
+                author_id=str(author_id),
+            )
+            return
+
+        result = await self.session.execute(
+            select(JournalTable).where(JournalTable.agent_id == author_id)
+        )
+        journal = result.scalar_one_or_none()
+        if journal is None:
+            journal = JournalTable(agent_id=author_id)
+            self.session.add(journal)
+            await self.session.flush()
+        self.session.add(
+            JournalEntryTable(
+                journal_id=journal.id,
+                type=JournalEntryType.DECISION_LOG,
+                title=title,
+                content=content,
+                task_id=task_id,
+            )
+        )
+
     async def ceo_reject(
         self,
         task_id: UUID,
@@ -3862,20 +3908,47 @@ class TaskService(BaseService):
         # Validate transition with CEO role requirement
         self._validate_and_set_status(task, TaskStatus.NEEDS_REVISION, "ceo")
 
-        # Try to reassign to original developer
+        # Surface the CEO's required changes through the task journal — the one
-        original_dev = extract_original_developer(task.quick_context)
+        # channel a downstream worker actually reads (evidence.journal_highlights
-        if original_dev:
+        # serves handoff entries into the briefing). quick_context above is
-            task.assigned_to = cast("Any", UUID(original_dev))
+        # write-only, so the reason would otherwise never reach the reworker.
-            task.claimed_by = cast("Any", UUID(original_dev))
+        await self._write_handoff_journal(
+            author_id=UUID(AGENT_UUIDS["ceo"]),
+            task_id=task_id,
+            title="CEO change request",
+            content=reason,
+        )
+
+        # Route the rejected task to whoever should drive the rework.
+        reassigned_to: str | None
+        if task.project_id is None and task.product_id is not None:
+            # Coordination/integration root: the Main PM delegates the rework — a
+            # board/dev role cannot drive a coordination task.
+            main_pm_id = UUID(AGENT_UUIDS["main-pm"])
+            task.team = Team.MAIN_PM
+            task.assigned_to = cast("Any", main_pm_id)
+            task.claimed_by = cast("Any", main_pm_id)
+            reassigned_to = str(main_pm_id)
             self.log.info(
-                "Task reassigned to original developer after CEO rejection",
+                "Coordination task rejected by CEO - routed to Main PM",
                 task_id=str(task_id),
-                original_developer=original_dev,
             )
         else:
-            # Clear assignment so it can be claimed
+            original_dev = extract_original_developer(task.quick_context)
-            task.assigned_to = None
+            if original_dev:
-            task.claimed_by = None
+                task.assigned_to = cast("Any", UUID(original_dev))
+                task.claimed_by = cast("Any", UUID(original_dev))
+                self.log.info(
+                    "Task reassigned to original developer after CEO rejection",
+                    task_id=str(task_id),
+                    original_developer=original_dev,
+                )
+                reassigned_to = original_dev
+            else:
+                # No tracked developer: leave for the pool to claim.
+                task.assigned_to = None
+                task.claimed_by = None
+                reassigned_to = None
 
         await self.session.flush()
 
@@ -3883,7 +3956,7 @@ class TaskService(BaseService):
         await self._emit_task_event(
             EventType.TASK_CEO_REJECTED,
             task_id,
-            {"reason": reason, "reassigned_to": original_dev},
+            {"reason": reason, "reassigned_to": reassigned_to},
         )
 
         self.log.info(

tests/integration/test_evidence_repo_queries.py

@@ -0,0 +1,167 @@
+"""EvidenceRepo real-DB query coverage.
+
+Exercises the actual SQL (array ``contains``, the a2a slug ``or_``, team+status
+filters) against a live Postgres — the part the mocked unit tests can't catch.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+import pytest_asyncio
+from roboco.db.tables import (
+    A2AConversationTable,
+    AgentTable,
+    NotificationTable,
+    ProjectTable,
+)
+from roboco.models import AgentRole, AgentStatus, Team
+from roboco.models.base import (
+    Complexity,
+    NotificationPriority,
+    NotificationType,
+    TaskNature,
+    TaskStatus,
+    TaskType,
+)
+from roboco.models.task import TaskCreateRequest
+from roboco.services.gateway.evidence_repo import EvidenceRepo
+from roboco.services.task import TaskService
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+@pytest_asyncio.fixture
+async def setup(db_session: AsyncSession) -> AsyncIterator[dict]:
+    agent = AgentTable(
+        id=uuid4(),
+        name="BE 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(agent)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="EV-Proj",
+        slug=f"ev-proj-{uuid4().hex[:8]}",
+        git_url="https://example.com/r.git",
+        assigned_cell=Team.BACKEND,
+        created_by=agent.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+    yield {
+        "repo": EvidenceRepo(db_session),
+        "svc": TaskService(db_session),
+        "agent": agent,
+        "project_id": project.id,
+        "db": db_session,
+    }
+
+
+@pytest.mark.asyncio
+async def test_pending_notifications_returns_unacked_for_agent(setup: dict) -> None:
+    agent = setup["agent"]
+    setup["db"].add(
+        NotificationTable(
+            type=NotificationType.ALERT,
+            priority=NotificationPriority.HIGH,
+            from_agent=agent.id,
+            to_agents=[agent.id],
+            subject="CEO change request",
+            body="redo the API contract",
+            timestamp=datetime.now(UTC),
+        )
+    )
+    await setup["db"].flush()
+    out = await setup["repo"].list_pending_notifications(agent.id)
+    assert len(out) == 1
+    assert out[0]["subject"] == "CEO change request"
+    # An unrelated agent sees nothing (array membership filter works).
+    assert await setup["repo"].list_pending_notifications(uuid4()) == []
+
+
+@pytest.mark.asyncio
+async def test_unread_a2a_returns_conversations_with_unread(setup: dict) -> None:
+    agent = setup["agent"]
+    now = datetime.now(UTC)
+    seeded_unread = 2
+    setup["db"].add(
+        A2AConversationTable(
+            agent_a=agent.slug,
+            agent_b="main-pm",
+            unread_by_a=seeded_unread,
+            unread_by_b=0,
+            topic="rework",
+            created_at=now,
+            updated_at=now,
+        )
+    )
+    await setup["db"].flush()
+    out = await setup["repo"].list_unread_a2a(agent.id)
+    assert len(out) == 1
+    assert out[0]["from_agent"] == "main-pm"
+    assert out[0]["unread"] == seeded_unread
+
+
+@pytest.mark.asyncio
+async def test_blockers_and_recent_activity_scoped_to_team(setup: dict) -> None:
+    svc, agent = setup["svc"], setup["agent"]
+    task = await svc.create(
+        TaskCreateRequest(
+            title="blocked thing",
+            description="d",
+            acceptance_criteria=["ac"],
+            team=Team.BACKEND,
+            created_by=agent.id,
+            project_id=setup["project_id"],
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            estimated_complexity=Complexity.MEDIUM,
+        )
+    )
+    task.status = TaskStatus.BLOCKED
+    await setup["db"].flush()
+
+    blockers = await setup["repo"].blockers_in_lane(agent.id)
+    assert any(b["task_id"] == str(task.id) for b in blockers)
+    recent = await setup["repo"].recent_team_activity(agent.id)
+    assert any(r["task_id"] == str(task.id) for r in recent)
+
+
+@pytest.mark.asyncio
+async def test_task_metadata_gaps_flags_missing(setup: dict) -> None:
+    svc, agent = setup["svc"], setup["agent"]
+    task = await svc.create(
+        TaskCreateRequest(
+            title="thin",
+            description="d",
+            acceptance_criteria=["ac"],
+            team=Team.BACKEND,
+            created_by=agent.id,
+            project_id=setup["project_id"],
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            estimated_complexity=Complexity.MEDIUM,
+        )
+    )
+    task.acceptance_criteria = []
+    task.description = ""
+    await setup["db"].flush()
+    gaps = await setup["repo"].task_metadata_gaps(task.id)
+    assert "no acceptance criteria" in gaps
+    assert "no description" in gaps

tests/integration/test_task_service_transitions.py

@@ -9,23 +9,31 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from unittest.mock import AsyncMock
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 import pytest_asyncio
-from roboco.db.tables import AgentTable, ProjectTable
+from roboco.db.tables import (
+    AgentTable,
+    JournalEntryTable,
+    ProductTable,
+    ProjectTable,
+)
 from roboco.events import EventType
 from roboco.models import AgentRole, AgentStatus, Team
 from roboco.models.base import (
     BlockerResolverType,
     Complexity,
+    JournalEntryType,
     TaskNature,
     TaskStatus,
     TaskType,
 )
 from roboco.models.task import TaskCreateRequest
+from roboco.seeds.initial_data import AGENT_UUIDS
 from roboco.services.base import NotFoundError
 from roboco.services.task import SoftBlockInfo, TaskService
+from sqlalchemy import select
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
@@ -577,6 +585,99 @@ async def test_ceo_reject_clears_assignment_when_no_original_dev(
     assert rejected.assigned_to is None
 
 
+@pytest.mark.asyncio
+async def test_ceo_reject_routes_coordination_task_to_main_pm(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """A rejected coordination/integration root goes to the Main PM to delegate,
+    not back to a developer."""
+    svc = task_setup["svc"]
+    main_pm_id = UUID(AGENT_UUIDS["main-pm"])
+    if await db_session.get(AgentTable, main_pm_id) is None:
+        db_session.add(
+            AgentTable(
+                id=main_pm_id,
+                name="Main PM",
+                slug="main-pm",
+                role=AgentRole.MAIN_PM,
+                team=Team.MAIN_PM,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="pm",
+                capabilities=[],
+                permissions={},
+                metrics={},
+            )
+        )
+    product = ProductTable(
+        name="P", slug=f"p-{uuid4().hex[:8]}", created_by=task_setup["agent_id"]
+    )
+    db_session.add(product)
+    await db_session.flush()
+
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.AWAITING_CEO_APPROVAL
+    task.project_id = None  # coordination root: no project, has product
+    task.product_id = product.id
+    await db_session.flush()
+
+    rejected = await svc.ceo_reject(task.id, reason="redo the API contract")
+    assert rejected is not None
+    assert rejected.status == TaskStatus.NEEDS_REVISION
+    assert rejected.team == Team.MAIN_PM
+    assert rejected.assigned_to == main_pm_id
+
+
+@pytest.mark.asyncio
+async def test_ceo_reject_writes_handoff_journal(
+    task_setup: dict, db_session: AsyncSession
+) -> None:
+    """The CEO's reason is recorded as a DECISION_LOG journal entry on the task —
+    the channel that actually reaches the reworker (quick_context does not)."""
+    svc = task_setup["svc"]
+    ceo_id = UUID(AGENT_UUIDS["ceo"])
+    if await db_session.get(AgentTable, ceo_id) is None:
+        db_session.add(
+            AgentTable(
+                id=ceo_id,
+                name="CEO",
+                slug="ceo",
+                role=AgentRole.CEO,
+                team=Team.MAIN_PM,
+                status=AgentStatus.ACTIVE,
+                model_config={},
+                system_prompt="ceo",
+                capabilities=[],
+                permissions={},
+                metrics={},
+            )
+        )
+        await db_session.flush()
+
+    task = await svc.create(_req(task_setup))
+    task.status = TaskStatus.AWAITING_CEO_APPROVAL
+    task.quick_context = f"original_developer:{task_setup['agent_id']}"
+    await db_session.flush()
+
+    reason = "AC9/AC10 totals must include cache tokens"
+    rejected = await svc.ceo_reject(task.id, reason=reason)
+    assert rejected is not None
+
+    entries = (
+        (
+            await db_session.execute(
+                select(JournalEntryTable).where(
+                    JournalEntryTable.task_id == task.id,
+                    JournalEntryTable.type == JournalEntryType.DECISION_LOG,
+                )
+            )
+        )
+        .scalars()
+        .all()
+    )
+    assert any(reason in (e.content or "") for e in entries)
+
+
 # ---------------------------------------------------------------------------
 # escalate_to_ceo - all error branches
 # ---------------------------------------------------------------------------

tests/unit/gateway/test_evidence_repo.py

@@ -1,8 +1,8 @@
-"""roboco.services.gateway.evidence_repo coverage.
+"""roboco.services.gateway.evidence_repo coverage (mapping + empty paths).
 
-Most methods are still Phase 1 stubs returning empty lists;
+These are mocked unit tests for the row-mapping and empty/not-found paths. The
-``journal_highlights_for_task`` is wired to a real query, so it is tested
+actual SQL (WHERE clauses, array operators) is exercised against a real DB in
-against a mocked ``execute`` result that stands in for the DB rows.
+tests/integration/test_evidence_repo_queries.py.
 """
 
 from __future__ import annotations
@@ -16,69 +16,118 @@ import pytest
 from roboco.services.gateway.evidence_repo import EvidenceRepo
 
 
-@pytest.fixture
+def _empty_repo() -> EvidenceRepo:
-def repo() -> EvidenceRepo:
+    """Repo whose scalar() and execute() both yield nothing."""
-    """Build an EvidenceRepo with a stub session — no DB access in Phase 1."""
-    return EvidenceRepo(MagicMock())
-
-
-@pytest.mark.asyncio
-async def test_constructor_stores_db_session() -> None:
-    fake_db = MagicMock()
-    repo = EvidenceRepo(fake_db)
-    assert repo._db is fake_db
-
-
-@pytest.mark.asyncio
-async def test_list_unread_a2a_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.list_unread_a2a(uuid4())
-    assert out == []
-
-
-@pytest.mark.asyncio
-async def test_list_unread_mentions_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.list_unread_mentions(uuid4())
-    assert out == []
-
-
-@pytest.mark.asyncio
-async def test_list_pending_notifications_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.list_pending_notifications(uuid4())
-    assert out == []
-
-
-@pytest.mark.asyncio
-async def test_task_metadata_gaps_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.task_metadata_gaps(uuid4())
-    assert out == []
-
-
-@pytest.mark.asyncio
-async def test_recent_team_activity_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.recent_team_activity(uuid4())
-    assert out == []
-
-
-@pytest.mark.asyncio
-async def test_blockers_in_lane_returns_empty(repo: EvidenceRepo) -> None:
-    out = await repo.blockers_in_lane(uuid4())
-    assert out == []
-
-
-def _repo_with_rows(rows: list[object]) -> EvidenceRepo:
-    """EvidenceRepo whose db.execute() yields a result with the given rows."""
     db = MagicMock()
+    db.scalar = AsyncMock(return_value=None)
+    result = MagicMock()
+    result.all.return_value = []
+    result.scalars.return_value.all.return_value = []
+    db.execute = AsyncMock(return_value=result)
+    return EvidenceRepo(db)
+
+
+def _repo_with_rows(rows: list[object], *, scalar: object = None) -> EvidenceRepo:
+    """Repo whose execute().all()/scalars().all() yields rows; scalar() → scalar."""
+    db = MagicMock()
+    db.scalar = AsyncMock(return_value=scalar)
     result = MagicMock()
     result.all.return_value = rows
+    result.scalars.return_value.all.return_value = rows
     db.execute = AsyncMock(return_value=result)
     return EvidenceRepo(db)
 
 
+@pytest.mark.asyncio
+async def test_constructor_stores_db_session() -> None:
+    fake_db = MagicMock()
+    assert EvidenceRepo(fake_db)._db is fake_db
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "method",
+    [
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+    ],
+)
+async def test_methods_return_empty_when_no_data(method: str) -> None:
+    repo = _empty_repo()
+    assert await getattr(repo, method)(uuid4()) == []
+
+
+@pytest.mark.asyncio
+async def test_pending_notifications_maps_rows() -> None:
+    ts = datetime(2026, 6, 9, 19, 27, tzinfo=UTC)
+    nid, frm, tid = uuid4(), uuid4(), uuid4()
+    row = SimpleNamespace(
+        id=nid,
+        type="alert",
+        priority="high",
+        subject="CEO change request",
+        body="redo the contract",
+        from_agent=frm,
+        related_task_id=tid,
+        timestamp=ts,
+    )
+    out = await _repo_with_rows([row]).list_pending_notifications(uuid4())
+    assert out == [
+        {
+            "notification_id": str(nid),
+            "type": "alert",
+            "priority": "high",
+            "subject": "CEO change request",
+            "body": "redo the contract",
+            "from_agent": str(frm),
+            "task_id": str(tid),
+            "timestamp": ts.isoformat(),
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_unread_a2a_maps_other_agent_and_unread_count() -> None:
+    cid = uuid4()
+    conv = SimpleNamespace(
+        id=cid,
+        agent_a="be-pm",
+        agent_b="main-pm",
+        unread_by_a=3,
+        unread_by_b=0,
+        topic="rework",
+        task_id=None,
+    )
+    out = await _repo_with_rows([conv], scalar="be-pm").list_unread_a2a(uuid4())
+    assert out == [
+        {
+            "conversation_id": str(cid),
+            "from_agent": "main-pm",
+            "unread": 3,
+            "topic": "rework",
+            "task_id": None,
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_task_metadata_gaps_flags_missing_fields() -> None:
+    repo = _repo_with_rows(
+        [], scalar=SimpleNamespace(acceptance_criteria=[], description="")
+    )
+    assert await repo.task_metadata_gaps(uuid4()) == [
+        "no acceptance criteria",
+        "no description",
+    ]
+
+
 @pytest.mark.asyncio
 async def test_journal_highlights_for_task_empty_when_no_entries() -> None:
-    repo = _repo_with_rows([])
+    assert await _repo_with_rows([]).journal_highlights_for_task(uuid4()) == []
-    out = await repo.journal_highlights_for_task(uuid4())
-    assert out == []
 
 
 @pytest.mark.asyncio
@@ -92,8 +141,7 @@ async def test_journal_highlights_for_task_maps_rows_with_author() -> None:
         slug="product-owner",
         role="product_owner",
     )
-    repo = _repo_with_rows([row])
+    out = await _repo_with_rows([row]).journal_highlights_for_task(uuid4())
-    out = await repo.journal_highlights_for_task(uuid4())
     assert out == [
         {
             "author": "product-owner",