"""Shared list-query grammar (§16.2) over read-model records. One grammar across ``GET /v1/messages``, ``/v1/results``, ``/v1/agents`` and ``/v1/inbox/{handle}``: filename-tier filters (``agent``, ``since``/``until``) prune before any content is touched; frontmatter/content filters run over the cached records; then order → cursor/limit → expand. """ from __future__ import annotations import re from datetime import datetime, timezone from app.errors import InvalidQuery from app.models import MessageListing, MessageRecord from app.naming import agent_from_filename from app.read_model import Record from app.validation import validate_agent_id STAMP_LEN = len("YYYYMMDD-HHmmss-mmm") _COMPACT_RE = re.compile(r"^\d{8}(?:-\d{6}(?:-\d{3})?)?$") VERIFICATION_STATES = ("pending", "valid", "invalid") def filename_stamp(filename: str) -> str: """The server-stamped chronological prefix of a message/result filename.""" return filename[:STAMP_LEN] def normalize_stamp(value: str, *, param: str) -> str: """Accept ISO 8601 or compact ``YYYYMMDD[-HHmmss[-mmm]]``; return a compact stamp comparable against the server-stamped filename prefix (UTC).""" v = value.strip() if _COMPACT_RE.match(v): if len(v) == 8: return v + "-000000-000" if len(v) == 15: return v + "-000" return v try: dt = datetime.fromisoformat(v.replace("Z", "+00:00")) except ValueError: raise InvalidQuery( f"`{param}` must be ISO 8601 or YYYYMMDD-HHmmss[-mmm], got {value!r}" ) if dt.tzinfo is not None: dt = dt.astimezone(timezone.utc) return dt.strftime("%Y%m%d-%H%M%S-") + f"{dt.microsecond // 1000:03d}" def parse_verification_param(value: str | None) -> set[str] | None: """CSV of verification states, e.g. ``valid,pending``. None → no filter.""" if value is None: return None states = {s.strip() for s in value.split(",") if s.strip()} bad = states - set(VERIFICATION_STATES) if bad or not states: raise InvalidQuery( f"`verification` must be a CSV of {VERIFICATION_STATES}, got {value!r}" ) return states def apply_filters( records: list[Record], *, agent: str | None = None, since: str | None = None, until: str | None = None, fm_eq: dict[str, str] | None = None, q: str | None = None, ) -> list[Record]: """``agent``/``since``/``until`` are answerable from filenames alone; ``fm_eq`` matches frontmatter values by string equality; ``q`` is a case-insensitive substring over frontmatter+body.""" out: list[Record] = [] for r in records: if agent is not None and agent_from_filename(r.filename) != agent: continue if since is not None and filename_stamp(r.filename) < since: continue if until is not None and filename_stamp(r.filename) > until: continue if fm_eq is not None: if any(str(r.frontmatter.get(k, "")) != want for k, want in fm_eq.items()): continue if q is not None and not _q_match(r, q): continue out.append(r) return out def _q_match(r: Record, q: str) -> bool: ql = q.lower() if ql in r.body.lower(): return True return any(ql in f"{k}: {v}".lower() for k, v in r.frontmatter.items()) def effective_limit(limit: int | None, expand: bool, cap: int) -> int | None: """Expanded pages are capped so one call can't serialize the whole corpus.""" if not expand: return limit if limit is None or limit <= 0 or limit > cap: return cap return limit def paginate( records: list[Record], *, order: str, limit: int | None, after: str | None, before: str | None, ) -> tuple[list[Record], str | None]: """Cursor + slice over filtered records (ascending filename order in). ``after``/``before`` are exclusive filename bounds. Returns the page and a ``next`` cursor (the page's last filename) when more matches remain in the traversal direction — pass it back as ``after`` for asc, ``before`` for desc. """ if after is not None: records = [r for r in records if r.filename > after] if before is not None: records = [r for r in records if r.filename < before] ordered = list(reversed(records)) if order == "desc" else records if limit is not None and 0 < limit < len(ordered): page = ordered[:limit] return page, page[-1].filename return ordered, None def list_message_like( records: list[Record], *, agent: str | None, since: str | None, until: str | None, type_: str | None, via: str | None, q: str | None, expand: bool, limit: int | None, order: str, after: str | None, before: str | None, expand_cap: int, ) -> MessageListing: """The full §16.2 pipeline for message-shaped folders (board and inboxes).""" if agent is not None: validate_agent_id(agent) fm_eq: dict[str, str] = {} if type_ is not None: fm_eq["type"] = type_ if via is not None: fm_eq["via"] = via filtered = apply_filters( records, agent=agent, since=normalize_stamp(since, param="since") if since is not None else None, until=normalize_stamp(until, param="until") if until is not None else None, fm_eq=fm_eq or None, q=q, ) page, next_cursor = paginate( filtered, order="desc" if order == "desc" else "asc", limit=effective_limit(limit, expand, expand_cap), after=after, before=before, ) items: list[str] | list[MessageRecord] if expand: items = [ MessageRecord( filename=r.filename, frontmatter=r.frontmatter, body=r.body, reasons=r.reasons, ) for r in page ] else: items = [r.filename for r in page] return MessageListing( count=len(records), matched=len(filtered), items=items, next=next_cursor, # The cursor to persist verbatim (WATCH_DESIGN.md §4.4): the newest # filename ON THIS PAGE, computed here so no client ever has to scan # author-controlled record content for a maximum. Independent of # `order` (asc pages end on it, desc pages start on it) and of `next` # (which is a pagination handle, not a read position). cursor=max((r.filename for r in page), default=None), )