Paginate C-25 owner-return reply collection (STATE-WP-0094-T03)

GET /messages/ accepts offset with stable created_at,id ordering; C-25 pages
every reply per counterpart, dedups by message id and warns explicitly when
collection is incomplete instead of capping at 200.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 671493@bnt-lap001
Assistant-Session: 60d7b11f-78f9-4731-9367-918511b2762d

api/routers/messages.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timezone
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response, status
+from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, Response, status
 from sqlalchemy import and_, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -137,6 +137,7 @@ async def list_messages(
     from_agent: str | None = None,
     unread_only: bool = False,
     limit: int = 50,
+    offset: int = Query(0, ge=0),
     session: AsyncSession = Depends(get_session),
 ) -> list[MessageRead]:
     now = utcnow()
@@ -170,7 +171,10 @@ async def list_messages(
         values = resolution.slug_values if resolution else (from_agent,)
         query = query.where(AgentMessage.from_agent.in_(values))
     result = await session.execute(
-        query.order_by(AgentMessage.created_at.desc()).limit(limit)
+        # ``id`` breaks created_at ties so offset pages are stable (STATE-WP-0094-T03).
+        query.order_by(AgentMessage.created_at.desc(), AgentMessage.id.desc())
+        .offset(offset)
+        .limit(limit)
     )
     messages = list(result.scalars().all())
     if reader_values is None or reader is None:

scripts/consistency_check.py

@@ -2786,6 +2786,48 @@ def collect_inbox_hygiene(api_base: str, repo_slug: str) -> dict[str, Any]:
     }
 
 
+C25_PAGE_SIZE = 200
+C25_MAX_PAGES = 50
+
+
+def _collect_counterpart_messages(
+    api_base: str, repo_slug: str, counterpart: str
+) -> tuple[list[dict], bool]:
+    """Page through every reply from ``counterpart`` (STATE-WP-0094-T03).
+
+    Returns ``(messages, complete)``. Messages are deduplicated by id, so a
+    reply shifting between pages while new mail arrives is counted once and the
+    finding's message ids stay stable. ``complete`` is False when the hub
+    ignored ``offset`` (a page repeated only known ids) or the page bound was
+    hit; the caller reports that instead of silently capping.
+    """
+    seen: dict[str, dict] = {}
+    anonymous: list[dict] = []
+    for page in range(C25_MAX_PAGES):
+        rows = _api_get(api_base, "/messages", {
+            "to_agent": repo_slug, "from_agent": counterpart,
+            "unread_only": False, "limit": C25_PAGE_SIZE, "offset": page * C25_PAGE_SIZE,
+        }) or []
+        if not isinstance(rows, list):
+            rows = []
+        new = 0
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            message_id = str(row.get("id") or "")
+            if not message_id:
+                anonymous.append(row)
+                new += 1
+            elif message_id not in seen:
+                seen[message_id] = row
+                new += 1
+        if len(rows) < C25_PAGE_SIZE:
+            return list(seen.values()) + anonymous, True
+        if new == 0:
+            return list(seen.values()) + anonymous, False
+    return list(seen.values()) + anonymous, False
+
+
 def _check_blocked_unblock_sweep(
     api_base: str,
     repo_slug: str,
@@ -2846,11 +2888,18 @@ def _check_blocked_unblock_sweep(
             if not counterpart:
                 continue
             if counterpart not in messages_by_agent:
-                rows = _api_get(api_base, "/messages", {
-                    "to_agent": repo_slug, "from_agent": counterpart,
-                    "unread_only": False, "limit": 200,
-                }) or []
-                messages_by_agent[counterpart] = rows if isinstance(rows, list) else []
+                rows, complete = _collect_counterpart_messages(api_base, repo_slug, counterpart)
+                messages_by_agent[counterpart] = rows
+                if not complete:
+                    report.add(
+                        severity="WARN", check_id="C-25",
+                        message=(f"Owner-return reply collection from {counterpart} is incomplete "
+                                 f"({len(rows)} message(s) read; hub ignored paging or the "
+                                 f"{C25_MAX_PAGES}-page bound was reached) — returns may be missing"),
+                        file_path=workplan_display_path(repo_dir, wp_file),
+                        file_value=f"message-from:{counterpart}",
+                        fixable=False,
+                    )
             # Dated notes are the available acknowledgement watermark. Date-only
             # notes cover the whole UTC day, avoiding same-day re-notification.
             note_dates = []

tests/test_c25_message_pagination.py

@@ -0,0 +1,94 @@
+"""C-25 owner-return collection pages past 200 replies (STATE-WP-0094-T03)."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
+
+from consistency_check import ConsistencyReport, _check_blocked_unblock_sweep  # noqa: E402
+
+BODY = (
+    "```task\nid: DEMO-WP-0001-T01\nstatus: wait\nblocked_on: message-from:provider\n```\n\n"
+    "2026-10-01: waiting on provider."
+)
+
+
+def _replies(total: int) -> list[dict]:
+    # Newest first, as the hub orders them.
+    return [
+        {"id": f"reply-{i:04d}", "from_agent": "provider",
+         "created_at": f"2026-10-03T{(i // 60) % 24:02d}:{i % 60:02d}:00Z", "read_at": None}
+        for i in range(total)
+    ][::-1]
+
+
+def _run(tmp_path, monkeypatch, fake_get):
+    monkeypatch.setattr("consistency_check._api_get", fake_get)
+    report = ConsistencyReport(repo_slug="demo", repo_path=str(tmp_path))
+    _check_blocked_unblock_sweep(
+        "http://unused", "demo", tmp_path,
+        [(tmp_path / "plan.md", {"status": "active", "flavor": "implementation"}, BODY)], report,
+    )
+    return report
+
+
+def test_collects_every_reply_across_pages_with_stable_ids(tmp_path, monkeypatch):
+    replies = _replies(450)
+    queries = []
+
+    def paged(_api_base, _path, params):
+        queries.append(dict(params))
+        start = params["offset"]
+        return replies[start:start + params["limit"]]
+
+    report = _run(tmp_path, monkeypatch, paged)
+    assert [q["offset"] for q in queries] == [0, 200, 400]
+    assert len(report.issues) == 1
+    ids = report.issues[0]._fix_context["message_ids"]
+    assert len(ids) == len(set(ids)) == 450
+    assert ids == sorted(ids)  # chronological, so the dedup key is stable
+    assert report.issues[0]._fix_context["task_record_id"] == "DEMO-WP-0001-T01"
+
+    # Overlapping pages (new mail shifting offsets) are deduplicated by id.
+    shifted = [{"id": "reply-new", "from_agent": "provider",
+                "created_at": "2026-10-03T23:59:00Z", "read_at": None}] + replies
+
+    def shifting(_api_base, _path, params):
+        source = replies if params["offset"] == 0 else shifted
+        return source[params["offset"]:params["offset"] + params["limit"]]
+
+    again = _run(tmp_path, monkeypatch, shifting)
+    assert len(again.issues[0]._fix_context["message_ids"]) == 450
+
+
+def test_hub_ignoring_offset_is_reported_not_silently_capped(tmp_path, monkeypatch):
+    replies = _replies(200)
+    report = _run(tmp_path, monkeypatch, lambda _a, _p, params: replies)
+    checks = [(issue.check_id, "incomplete" in issue.message) for issue in report.issues]
+    assert ("C-25", True) in checks
+    owner = [i for i in report.issues if i._fix_context.get("task_record_id")]
+    assert len(owner[0]._fix_context["message_ids"]) == 200
+
+
+@pytest.mark.asyncio
+async def test_messages_api_offset_pages_are_disjoint_and_ordered(client):
+    for index in range(5):
+        response = await client.post("/messages/", json={
+            "from_agent": "provider", "to_agent": "demo",
+            "subject": f"reply {index}", "body": "done",
+        })
+        assert response.status_code == 201, response.text
+    pages = []
+    for offset in (0, 2, 4):
+        response = await client.get("/messages/", params={
+            "to_agent": "demo", "from_agent": "provider", "limit": 2, "offset": offset})
+        assert response.status_code == 200, response.text
+        pages.append([row["id"] for row in response.json()])
+    flat = [item for page in pages for item in page]
+    assert [len(page) for page in pages] == [2, 2, 1]
+    assert len(set(flat)) == 5
+    bad = await client.get("/messages/", params={"offset": -1})
+    assert bad.status_code == 422