diff --git a/api/routers/messages.py b/api/routers/messages.py index cfe102c..eba28ea 100644 --- a/api/routers/messages.py +++ b/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: diff --git a/scripts/consistency_check.py b/scripts/consistency_check.py index eb581df..11b6bc6 100644 --- a/scripts/consistency_check.py +++ b/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 = [] diff --git a/tests/test_c25_message_pagination.py b/tests/test_c25_message_pagination.py new file mode 100644 index 0000000..39b156d --- /dev/null +++ b/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