"""Assistant proposal surfaces: a read-only page, and the admin audit views. **The confirmation page cannot submit anything, by construction.** That is the fix for a class of defect four consecutive review rounds failed to patch by adjustment. The page opens from a Feishu card, a Feishu card is forwardable, and the signed token travels inside it — so every attempt to decide "may THIS reader act?" from a web request collided with the fact that the Feishu in-app browser cannot reliably hold a Helix session (``client.ts`` records the iOS webview dropping the refresh cookie in production). Each round moved the compromise and the next found the new edge: a login wall that stranded the only population the page serves, then an open page with an enabled button over a guaranteed 401, then a button nobody — including the owner — could reach at all. Submitting belongs to the card's native ``card.action.trigger`` button, where Feishu hands us a signature-verified operator identity. That cannot be forwarded, needs no browser, no session and no SSO, and it is the primary button on every confirmation card the assistant sends (``confirm_card._prepend_native_button``, wired from both ``propose_tools`` and ``escalation``). Deleting the web submit path removes the whole question rather than answering it a fifth time. What remains is the one job the page was created for: the card truncates a value at 120 characters (``confirm_card.clip_value``) and this shows the rest. It stays public — it must open with no session — and a reader who is not signed in as the proposal's own person gets the same truncation the card applied, and never the execution ``result``. """ from __future__ import annotations import logging import uuid from datetime import datetime from typing import Any from fastapi import APIRouter, Body, Depends, Query, Request from sqlalchemy.ext.asyncio import AsyncSession from app.assistant import admin_service from app.assistant.actions import get_action from app.assistant.proposals import ( ProposalError, expire_if_lapsed, load_for_token, ) from app.assistant.settings import assistant_enabled from app.auth.models import AuthUser from app.core.coordinators.assistant_outcome_coordinator import ( list_submissions_for_person, ) from app.core.database import get_db from app.core.dependencies import ( bearer_scheme, get_current_user, require_roles, ) from app.core.locale import DEFAULT_LANG, Lang, normalize_lang from app.core.rbac import AUDITOR_ROLE, GLOBAL_ADMIN_ROLE from app.core.response import paginated_response, success_response logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/assistant", tags=["assistant"]) #: Deliberately narrower than ``require_admin_read``. Transcripts contain #: employees' own words about their problems, which is a step more sensitive #: than the governance data the leadership read tier normally sees — so #: ``management`` is not admitted here. The IT owner and the auditor are the #: two roles with an actual need. require_assistant_audit = require_roles(GLOBAL_ADMIN_ROLE, AUDITOR_ROLE) admin_router = APIRouter( prefix="/api/v1/assistant/admin", tags=["assistant-admin"], dependencies=[Depends(require_assistant_audit)], ) def _lang_of(raw: str | None) -> Lang: return normalize_lang(raw) or DEFAULT_LANG async def _optional_user( request: Request, db: AsyncSession = Depends(get_db), ) -> AuthUser | None: """The signed-in person, or ``None`` — never a 401. The confirm page must open for a reader with no Helix session (the Feishu in-app browser, where silent re-auth is known to fail on iOS), and simply show less. """ try: credentials = await bearer_scheme(request) if credentials is None: return None return await get_current_user(request, credentials, db) except Exception: # noqa: BLE001 — absent or unusable credential = anonymous return None def _clipped(row: dict[str, Any]) -> dict[str, Any]: """One summary row, truncated the way the card truncates it.""" from app.assistant.confirm_card import clip_value # noqa: PLC0415 return {**row, "value": clip_value(str(row.get("value") or ""))} async def _require_enabled(db: AsyncSession) -> None: if not await assistant_enabled(db): raise ProposalError( "ASSISTANT_DISABLED", "The assistant is not enabled.", status_code=404, ) @router.get("/proposals", summary="Show a pending assistant proposal") async def show_proposal( request: Request, token: str = Query(min_length=1, max_length=4096), lang: str | None = Query(default=None), viewer: AuthUser | None = Depends(_optional_user), db: AsyncSession = Depends(get_db), ): """Render-only view of the offer. Never consumes the token.""" await _require_enabled(db) proposal, _payload = await load_for_token(db, token) is_owner = viewer is not None and str(viewer.id) == str(proposal.person_id) # A lapsed proposal must read as lapsed: the page tells the employee # whether the card in their chat is still worth tapping. await expire_if_lapsed(db, proposal) action = get_action(proposal.action_key) if action is None: raise ProposalError( "ASSISTANT_ACTION_UNKNOWN", "This request can no longer be submitted.", status_code=410, ) resolved = _lang_of(lang) params = dict(proposal.params or {}) return success_response( data={ "action_key": proposal.action_key, "title": action.localized_title(resolved), "intent": action.localized_intent(resolved), # The owner sees everything; anyone else sees exactly what the card # showed. Same rows either way — only the long free text differs, # and that is the only thing the card was hiding. "rows": [ row.localized(resolved) if is_owner else _clipped(row.localized(resolved)) for row in action.summarise(params, resolved) ], "params_hash": proposal.params_hash, "status": proposal.status, # Ownership counts, not just status. Reopening GET to anonymous # readers without this put an ENABLED 确认提交 button in front of # someone whose POST is a guaranteed 401 — the exact shape the # comment 20 lines up calls out for lapsed proposals ("rendering an # enabled button on something the confirm call will refuse is how # an employee ends up staring at a raw backend error"), recreated # for unauthenticated ones by the fix that reopened the route. # No ``actionable`` / ``viewer_is_owner``: nothing here acts, so # there is no button state to describe. The page renders the # request and points at the card's own button. "expires_at": proposal.expires_at, # Never carried by the card, so a non-owner has no business # reading the ticket number back off a forwarded link. "result": proposal.result if is_owner else None, }, request=request, ) @router.post( "/proposals/{proposal_id}/confirm", summary="Confirm one of my own assistant offers", ) async def confirm_own_proposal( proposal_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Submit an offer the assistant made to the signed-in employee. **This is not the web submit path this module's docstring says was deleted.** That one was reached from a FORWARDABLE LINK carrying a signed token, opened in a browser that provably could not hold a Helix session, and so had no way to answer "may THIS reader act?". Every property that made it unanswerable is absent here: * no token — the caller names a proposal id and nothing else; * a real Helix session, because ``/me/support`` is behind one already, so the actor is the authenticated person rather than whoever opened a link; * ownership re-checked against that person inside ``confirm_and_execute_for_person``, which is the same function the Feishu card button calls — and a stricter caller than it, since that path takes the acting identity from callback body content. Idempotent by claim, not by check: the ``pending → confirmed`` update is conditional, so a double-click loses the race in the database and gets a 409 rather than submitting twice. ⚠️ **A personal access token is refused here**, and this is the one place in the platform where that is true. ``get_current_user`` accepts a PAT anywhere, which is right for every other mutation — a PAT is the employee delegating their own authority to a script. It is wrong for exactly this call: the entire point of a proposal is that a HUMAN read what would be submitted and said yes. An AI client holding the employee's PAT could otherwise ask the assistant for something and then approve it on the employee's behalf, which turns the confirmation step into a formality it performs against itself. Nothing in the assistant's own tool surface reaches this route (the read bridge is read-only and there is no confirm tool), so the refusal costs no shipped capability — it closes the path before something grows into it. """ await _require_enabled(db) # An ALLOWLIST, not a denylist — and that polarity is the whole point. # ``require_browser_session`` (core/dependencies.py) carries the scar: its # first shape tested ``auth_via == "pat"`` with a default of ``"jwt"``, # which meant any caller reaching the route without stamping the flag # sailed through the gate it was meant to fail closed. Written the same way # here, a future credential kind that does not set ``auth_via`` would be # able to confirm an offer on an employee's behalf. Only a browser session # gets through. if getattr(request.state, "auth_via", "") != "jwt": raise ProposalError( "ASSISTANT_PROPOSAL_NEEDS_A_PERSON", "Confirming an assistant offer needs a signed-in person, not a token.", status_code=403, ) from app.assistant.proposals import confirm_and_execute_for_person proposal = await confirm_and_execute_for_person( db, proposal_id=proposal_id, person=user, trace_id=str(getattr(request.state, "trace_id", "")), ) return success_response( data={ "id": str(proposal.id), "status": proposal.status, "action_key": proposal.action_key, # Only the owner ever reaches this line, so the execution result is # theirs to see. It is what turns "submitted" into a link to the # thing that was created. "result": proposal.result, }, request=request, ) @router.get("/submissions", summary="What the assistant submitted for me") async def list_my_submissions( request: Request, limit: int = Query(20, ge=1, le=100), # ``AuthUser`` is an alias of ``IdentityPerson`` (app/auth/models.py) — # same class, same id, so ``person.id`` downstream is the person id. user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Live state of everything this person asked the assistant to do. Read-only, self-scoped, and re-authorised per row: the query filters on the link's ``person_id``, and the resolver then re-derives permission from the domain row itself. Only the second survives an object changing hands, which is why both are here rather than one. Returns phases, not sentences. The reader's language is not known at this layer, and a rendered string would force every other surface to re-parse it. """ await _require_enabled(db) submissions = await list_submissions_for_person(db, person=user, limit=limit) # The resolver refreshes each link's funnel cache as a side effect of being # read; commit so that survives the request. await db.commit() return success_response( data=[s.as_dict() for s in submissions], request=request, ) @admin_router.get("/conversations", summary="List assistant conversations") async def admin_list_conversations( request: Request, person_id: uuid.UUID | None = Query(default=None), outcome: str | None = Query(default=None, max_length=32), tool: str | None = Query(default=None, max_length=64), chat_kind: str | None = Query( default=None, max_length=16, description="dm / handoff / group —— 私聊、Helix 建的工单群、其它群", ), since: datetime | None = Query(default=None), until: datetime | None = Query(default=None), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=admin_service.MAX_PAGE_SIZE), db: AsyncSession = Depends(get_db), ): rows, total = await admin_service.list_conversations( db, filters=admin_service.ConversationFilters( person_id=person_id, outcome=outcome, tool=tool, since=since, until=until, chat_kind=chat_kind, ), page=page, page_size=page_size, ) return paginated_response(rows, total, page, page_size, request) @admin_router.get( "/conversations/{conversation_id}", summary="Read one assistant transcript", ) async def admin_get_conversation( conversation_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db), ): found = await admin_service.get_conversation(db, conversation_id) if found is None: raise ProposalError( "ASSISTANT_CONVERSATION_NOT_FOUND", "No such conversation.", status_code=404, ) return success_response(data=found, request=request) @admin_router.get("/funnel", summary="Assistant submission funnel, end to end") async def admin_funnel( request: Request, days: int = Query(default=30, ge=1, le=365), db: AsyncSession = Depends(get_db), ): """Offered → pressed → executed → linked → what actually became of it. Separate from ``/stats`` on purpose: that one measures the conversation, this one measures whether anything happened. Two counts it keeps apart that the old numbers merged — ``closed_verified`` (a real object reached ``effective``) versus ``closed_on_word`` (the employee said it helped and there was nothing to verify), and ``pressed`` versus ``executed``. """ return success_response( data=await admin_service.submission_funnel(db, days=days), request=request, ) @admin_router.get("/stats", summary="Assistant usage and refusal counters") async def admin_stats( request: Request, days: int = Query(default=30, ge=1, le=365), db: AsyncSession = Depends(get_db), ): return success_response( data=await admin_service.usage_stats(db, days=days), request=request, ) @admin_router.get( "/unhelpful", summary="Answers employees marked as not helpful, with the question asked", ) async def admin_unhelpful( request: Request, days: int = Query(default=30, ge=1, le=365), limit: int = Query(default=50, ge=1, le=200), status: str | None = Query(default=None, max_length=20), db: AsyncSession = Depends(get_db), ): """The knowledge-gap queue. Deliberately paired with the question: a list of bad answers is a complaint, while question-and-answer together is something an IT admin can close by writing one document. """ return success_response( data=await admin_service.unhelpful_answers(db, days=days, limit=limit, status=status), request=request, ) @admin_router.post("/knowledge-gaps/{message_id}/transition") async def transition_knowledge_gap( message_id: uuid.UUID, request: Request, payload: dict[str, str | None] = Body(...), current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Claim, resolve, ignore, or reopen a gap. All transitions are idempotent.""" from app.assistant.gap_verification import GapClosureError action = str(payload.get("action") or "") try: result = await admin_service.update_gap( db, message_id=message_id, actor_id=current_user.id, action=action, note=payload.get("note"), fix_type=payload.get("fix_type"), evidence_ref=payload.get("evidence_ref"), trace_id=str(getattr(request.state, "trace_id", "")), ) except GapClosureError as exc: # Its own code, not the generic one: the UI has to tell "you have not # said what was wrong" apart from "the replay failed", and those lead # to different next actions. raise ProposalError(exc.code, str(exc), status_code=422) from exc except ValueError as exc: raise ProposalError("ASSISTANT_GAP_ACTION_INVALID", str(exc), status_code=422) from exc if result is None: raise ProposalError("ASSISTANT_GAP_NOT_FOUND", "No such knowledge gap.", status_code=404) return success_response(data=result, request=request) @admin_router.post("/knowledge-gaps/{message_id}/replay") async def replay_knowledge_gap( message_id: uuid.UUID, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Ask the gap's original question again, under the original asker's permissions, and record the shape of the turn. Does not close anything. A passing replay with no stated fix would let "it works now, somehow" count as a resolution — which is the shape of the problem this replaces, one level up. """ await _require_enabled(db) result = await admin_service.replay_gap( db, message_id=message_id, actor_id=current_user.id, trace_id=str(getattr(request.state, "trace_id", "")), ) if result is None: raise ProposalError( "ASSISTANT_GAP_NOT_REPLAYABLE", "This gap has no original question that can be replayed.", status_code=404, ) return success_response(data=result, request=request) @admin_router.get( "/knowledge-gaps/regression-set", summary="Questions that must keep passing", ) async def knowledge_gap_regression_set( request: Request, limit: int = Query(default=200, ge=1, le=500), db: AsyncSession = Depends(get_db), ): """Every question that failed, was fixed, and now passes. A set rather than a memory: the next change to retrieval or to a tool description can be checked against all of them. """ return success_response( data=await admin_service.regression_set(db, limit=limit), request=request, ) # ── 转人工路由(后台配置) ────────────────────────────────────────── #: 读:全局管理员 + 审计员(和这个 admin_router 的其余部分一致)。 #: 写:**只有**全局管理员 —— 审计员的职责是看,而改派负责人是一次运营决定。 require_route_write = require_roles(GLOBAL_ADMIN_ROLE) def _route_row( row: Any, owner: Any | None, backup: Any | None = None, ) -> dict[str, Any]: return { "id": str(row.id), "category_key": row.category_key, "display_name": row.display_name, "hint": row.hint, "owner_person_id": str(row.owner_person_id) if row.owner_person_id else None, # 名字给人读,id 给前端选择器用。名字**不入库** —— 它会变,而这一行会 # 活很久(见模型注释)。 "owner_name": ( (owner.display_name or owner.work_email) if owner is not None else None ), # 备份负责人:主负责人今天休假、或者压根联系不上时拉的那个人。 # 没有它的话,负责人休假两周意味着那两周里每一次转人工都把员工和一个 # 不会回话的人关进一个房间(见 ``owner_reach``)。 "backup_owner_person_id": ( str(row.backup_owner_person_id) if row.backup_owner_person_id else None ), "backup_owner_name": ( (backup.display_name or backup.work_email) if backup is not None else None ), "is_default": bool(row.is_default), "enabled": bool(row.enabled), "sort_order": int(row.sort_order), } @admin_router.get( "/escalation-routes", summary="转人工路由:哪一类问题找谁", ) async def list_escalation_routes( db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """全部路由(含未启用的)。 列出未启用的行是刻意的:一条被关掉的路由仍然是配置的一部分,而管理员要能 看到「安全这一类现在是关的」——否则它看起来像凭空消失。 """ from sqlalchemy import select # noqa: PLC0415 from app.assistant.models import AssistantEscalationRoute # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 rows = list((await db.execute( select(AssistantEscalationRoute).order_by( AssistantEscalationRoute.sort_order, AssistantEscalationRoute.category_key, ) )).scalars().all()) owner_ids = { pid for r in rows for pid in (r.owner_person_id, r.backup_owner_person_id) if pid } owners = {} if owner_ids: owners = { p.id: p for p in (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids)) )).scalars().all() } return success_response([ _route_row( r, owners.get(r.owner_person_id) if r.owner_person_id else None, owners.get(r.backup_owner_person_id) if r.backup_owner_person_id else None, ) for r in rows ]) @admin_router.put( "/escalation-routes/{route_id}", summary="改一条转人工路由", dependencies=[Depends(require_route_write)], ) async def update_escalation_route( route_id: uuid.UUID, payload: dict[str, Any] = Body(...), db: AsyncSession = Depends(get_db), user: AuthUser = Depends(get_current_user), ) -> dict[str, Any]: """改负责人 / 提示词 / 启用 / 兜底 / 排序。 ``category_key`` **不可改** —— 它是分类审计里存下来的稳定标识,改它会让 历史台账指向一个不存在的类别。要换语义就新建一条、关掉旧的。 """ from sqlalchemy import update # noqa: PLC0415 from app.assistant.models import AssistantEscalationRoute # noqa: PLC0415 from app.core.exceptions import AppException, NotFoundError # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 row = await db.get(AssistantEscalationRoute, route_id) if row is None: raise NotFoundError("AssistantEscalationRoute", str(route_id)) if "owner_person_id" in payload: raw = payload.get("owner_person_id") if raw in (None, ""): row.owner_person_id = None else: # 畸形 id 是 422,不是 500。``uuid.UUID()`` 对一个手打错的字符串抛 # ValueError,裸着让它冒出去会变成一个 500 —— 而这是配置面上一次 # 普通的输入错误,管理员该看到「这个 id 不对」而不是「服务器错误」。 try: owner_id = uuid.UUID(str(raw)) except (ValueError, AttributeError, TypeError) as exc: raise AppException( code="ASSISTANT_ROUTE_OWNER_INVALID", message=f"owner_person_id 不是一个合法的 id:{raw!r}", status_code=422, ) from exc owner = await db.get(IdentityPerson, owner_id) if owner is None: raise NotFoundError("IdentityPerson", str(raw)) row.owner_person_id = owner.id if "backup_owner_person_id" in payload: # 与主负责人同一套校验:畸形 id 是 422 而不是 500 —— 这是配置面上一次 # 普通的输入错误。 raw = payload.get("backup_owner_person_id") if raw in (None, ""): row.backup_owner_person_id = None else: try: backup_id = uuid.UUID(str(raw)) except (ValueError, AttributeError, TypeError) as exc: raise AppException( code="ASSISTANT_ROUTE_OWNER_INVALID", message=f"backup_owner_person_id 不是一个合法的 id:{raw!r}", status_code=422, ) from exc backup = await db.get(IdentityPerson, backup_id) if backup is None: raise NotFoundError("IdentityPerson", str(raw)) row.backup_owner_person_id = backup.id if "hint" in payload: hint = payload.get("hint") row.hint = str(hint)[:2000] if hint else None if "enabled" in payload: row.enabled = bool(payload.get("enabled")) if "sort_order" in payload: row.sort_order = int(payload.get("sort_order") or 100) if "display_name" in payload: # 三语齐全才接受:一个只有中文的显示名会让群名对英文读者变成乱码般的 # 存在,而群名是员工和负责人都看得见的东西。 # # 不齐全时**报错**,不静默忽略。原来是「凑不齐三语就跳过」,于是接口 # 返回 200 而那一行根本没改 —— 管理员以为自己保存了。一个说「已保存」 # 却没保存的配置面,正是这一轮在别处修的同一类谎。 raw_name = payload.get("display_name") supplied = ( { k: str(v) for k, v in raw_name.items() if k in ("zh", "en", "ja") and str(v).strip() } if isinstance(raw_name, dict) else {} ) if len(supplied) != 3: raise AppException( code="ASSISTANT_ROUTE_NAME_INCOMPLETE", message="display_name 需要同时给出 zh / en / ja 三种语言。", status_code=422, ) row.display_name = supplied if payload.get("is_default") is True: # 兜底路由恰好一条。先全置假再置真 —— 两条兜底会让 `_pick_route` 的 # 结果取决于查询顺序,那是最难查的一类不确定。 await db.execute( update(AssistantEscalationRoute).values(is_default=False), ) row.is_default = True await db.flush() owner = ( await db.get(IdentityPerson, row.owner_person_id) if row.owner_person_id else None ) backup = ( await db.get(IdentityPerson, row.backup_owner_person_id) if row.backup_owner_person_id else None ) from app.audit.service import AuditService # noqa: PLC0415 await AuditService(db).log( trace_id=uuid.uuid4().hex, actor_type="user", actor_user_id=user.id, action="assistant.escalation_route.update", status="success", resource_type="assistant_escalation_route", resource_id=str(row.id), detail={ "category_key": row.category_key, "owner_person_id": str(row.owner_person_id) if row.owner_person_id else None, "backup_owner_person_id": ( str(row.backup_owner_person_id) if row.backup_owner_person_id else None ), "enabled": row.enabled, "is_default": row.is_default, }, ) await db.commit() return success_response(_route_row(row, owner, backup)) @admin_router.get( "/room-console", summary="转人工与工单群运营台:概览 + 房间清单", ) async def room_console( request: Request, window_days: int = Query(30, ge=1, le=365), state: str | None = Query(default=None, max_length=16), ticket_open: bool | None = Query(default=None), opened_reason: str | None = Query(default=None, max_length=16), missing_record: bool | None = Query(default=None), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """这一页要回答的四个问题,一次请求一个快照。 **为什么概览和清单在同一个端点里**:四块数各自要一到三次聚合查询,而它们 必须是**同一时刻**的 —— 一个页面上写着「共 40 个群」而下面列出 37 行,读的 人会以为丢了三行,然后再也不信这一页。 """ from app.assistant import room_console as console # noqa: PLC0415 filters = console.ChatFilters( state=state, ticket_open=ticket_open, opened_reason=opened_reason, missing_record=missing_record, ) items, total = await console.list_chats( db, filters=filters, page=page, page_size=page_size, ) return success_response( data={ "overview": await console.overview(db, window_days=window_days), "unassigned_routes": await console.unassigned_routes(db), "chats": { "items": items, "total": total, "page": page, "page_size": page_size, }, }, request=request, ) @admin_router.get("/handoffs", summary="转人工台账:分类准不准、拉给了谁") async def list_handoffs( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """最近的转人工记录。 存在的理由是**评估路由质量**:模型判成什么类、置信度多少、有没有降级。 没有这张列表,「分类准不准」就只能靠人凭印象说。 """ from sqlalchemy import func, select # noqa: PLC0415 from app.assistant.models import AssistantHandoff # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 total = int((await db.execute( select(func.count()).select_from(AssistantHandoff) )).scalar() or 0) rows = list((await db.execute( select(AssistantHandoff) .order_by(AssistantHandoff.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) )).scalars().all()) people = {} if rows: people = { p.id: p for p in (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_({r.person_id for r in rows}), ) )).scalars().all() } items = [ { "id": str(r.id), "person_name": ( (people[r.person_id].display_name or people[r.person_id].work_email) if r.person_id in people else None ), "classified_category": r.classified_category, "classified_confidence": r.classified_confidence, "classified_reason": r.classified_reason, "chat_created": bool(r.feishu_chat_id), "degraded_reason": r.degraded_reason, "created_at": r.created_at.isoformat(), } for r in rows ] return paginated_response(items, total=total, page=page, page_size=page_size)