"""Read model behind the administrator's conversation-audit surface. The question this exists to answer is "what did the assistant tell this employee, and what did it look at to say that" — months later, without reconstructing anything. So the reads are shaped around a human investigation: find the conversation, open it, and see each answer next to the tools that produced it and the cost it incurred. Refused tool calls are surfaced as a first-class number rather than buried in logs. A refusal is the signal that something tried to reach past the read-only boundary, which is exactly the thing an administrator would want to notice early — a quiet week of zeros is meaningful, and a spike is actionable. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Select, func, select, text, tuple_ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from app.assistant.models import ( GAP_IGNORED, GAP_IN_PROGRESS, GAP_OPEN, GAP_RESOLVED, OUTCOME_ANSWERED, ROLE_ASSISTANT, ROLE_USER, VERDICT_HELPFUL, VERDICT_UNHELPFUL, AssistantConversation, AssistantMessage, AssistantMessageFeedback, ) from app.audit.models import AuditLog from app.audit.service import AuditService from app.identity.person_models import IdentityPerson ACTION_TURN = "assistant.message.received" logger = logging.getLogger(__name__) def _as_utc(value: datetime) -> datetime: """Convert an offset-aware bound to UTC; leave a naive one alone. The driver formats a datetime as ``YYYY-MM-DD HH:MM:SS`` and **ignores tzinfo entirely**, while these columns hold UTC wall clock. So an auditor scripting ``since=2026-08-01T00:00:00+08:00`` had their window silently shifted by eight hours — no error, no conversion, just conversations missing from the edges of the range they asked for. Naive input is taken as UTC, which is what the rest of the platform means by a bare timestamp. """ if value.tzinfo is None: return value return value.astimezone(UTC).replace(tzinfo=None) ACTION_TOOL_REFUSED = "assistant.tool.refused" #: Ceiling on the tool-usage scan behind ``usage_stats``. Well above a year of #: real traffic (production holds 78 assistant messages in total), so it changes #: nothing today — it exists so the page cannot become the slowest thing in the #: console as volume grows, and so the truncation is stated rather than assumed. _TOOL_SCAN_LIMIT = 20_000 ACTION_PROPOSAL_CONFIRMED = "assistant.proposal.confirmed" MAX_PAGE_SIZE = 100 @dataclass(frozen=True) class ConversationFilters: person_id: uuid.UUID | None = None outcome: str | None = None tool: str | None = None since: datetime | None = None until: datetime | None = None #: ``dm`` / ``handoff`` / ``group``。群里的一轮和私聊的一轮读起来不是一回事 #: (一个人看见 vs 一屋子人看见),混在一张表里翻是找不到的。 chat_kind: str | None = None def _escape_like(value: str) -> str: """Neutralise LIKE metacharacters in an operator-typed search string.""" return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def _apply_filters(stmt: Select, filters: ConversationFilters) -> Select: if filters.person_id is not None: stmt = stmt.where(AssistantConversation.person_id == filters.person_id) if filters.chat_kind: stmt = stmt.where(AssistantConversation.chat_kind == filters.chat_kind) if filters.since is not None: stmt = stmt.where( AssistantConversation.last_message_at >= _as_utc(filters.since), ) if filters.until is not None: stmt = stmt.where( AssistantConversation.last_message_at <= _as_utc(filters.until), ) if filters.outcome or filters.tool: # Correlated EXISTS rather than a join: a conversation with three # matching messages must appear once, not three times. inner = select(AssistantMessage.id).where( AssistantMessage.conversation_id == AssistantConversation.id, ) if filters.outcome: inner = inner.where(AssistantMessage.outcome == filters.outcome) if filters.tool: # ``json_search`` matches with LIKE semantics, so an unescaped # ``%`` or ``_`` in the search box would quietly become a wildcard. inner = inner.where( func.json_search( AssistantMessage.tool_calls, "one", _escape_like(filters.tool), "\\", ).is_not(None), ) stmt = stmt.where(inner.exists()) return stmt async def list_conversations( db: AsyncSession, *, filters: ConversationFilters, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: """Newest-first conversations with the person's name resolved.""" page_size = max(1, min(page_size, MAX_PAGE_SIZE)) page = max(1, page) total = await db.scalar( _apply_filters( select(func.count(AssistantConversation.id)), filters, ), ) stmt = ( _apply_filters( # OUTER join: the count above does not join at all, so an inner # join here would silently drop a conversation whose person row is # gone and leave the page showing 19 rows above a total of 21. select(AssistantConversation, IdentityPerson).outerjoin( IdentityPerson, IdentityPerson.id == AssistantConversation.person_id, ), filters, ) .order_by(AssistantConversation.last_message_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) rows = (await db.execute(stmt)).all() return ( [ { "id": str(conversation.id), "person_id": str(conversation.person_id), "person_name": person.display_name if person else None, "person_email": person.work_email if person else None, "channel": conversation.channel, "chat_kind": conversation.chat_kind, "chat_id": conversation.chat_id, "turn_count": conversation.turn_count, "started_at": conversation.started_at, "last_message_at": conversation.last_message_at, } for conversation, person in rows ], int(total or 0), ) async def get_conversation( db: AsyncSession, conversation_id: uuid.UUID, ) -> dict[str, Any] | None: """One transcript, in the order it happened, with each answer's evidence.""" conversation = await db.get(AssistantConversation, conversation_id) if conversation is None: return None person = await db.get(IdentityPerson, conversation.person_id) messages = ( ( await db.execute( select(AssistantMessage) .where(AssistantMessage.conversation_id == conversation_id) .order_by(AssistantMessage.seq), ) ) .scalars() .all() ) return { "id": str(conversation.id), "person_id": str(conversation.person_id), "person_name": person.display_name if person else None, "person_email": person.work_email if person else None, "channel": conversation.channel, "turn_count": conversation.turn_count, "started_at": conversation.started_at, "last_message_at": conversation.last_message_at, "messages": [ { "seq": m.seq, "role": m.role, "content": m.content, "tool_calls": m.tool_calls or [], # Captured from the KB tool's raw result at answer time. This # is evidence for the employee's real answer, not a fresh # re-search that may drift after the source document changes. "sources": m.sources or [], "outcome": m.outcome, "provider": m.provider, "latency_ms": m.latency_ms, "trace_id": m.trace_id, "created_at": m.created_at, } for m in messages ], } async def usage_stats(db: AsyncSession, *, days: int = 30) -> dict[str, Any]: """Operating picture for the window: volume, outcomes, tools, refusals.""" days = max(1, min(days, 365)) since = datetime.now(UTC) - timedelta(days=days) conversations = await db.scalar( select(func.count(AssistantConversation.id)).where( AssistantConversation.last_message_at >= since, ), ) people = await db.scalar( select(func.count(func.distinct(AssistantConversation.person_id))).where( AssistantConversation.last_message_at >= since, ), ) outcome_rows = ( await db.execute( select(AssistantMessage.outcome, func.count(AssistantMessage.id)) .where( AssistantMessage.created_at >= since, AssistantMessage.role == ROLE_ASSISTANT, ) .group_by(AssistantMessage.outcome), ) ).all() latency = await db.scalar( select(func.avg(AssistantMessage.latency_ms)).where( AssistantMessage.created_at >= since, AssistantMessage.role == ROLE_ASSISTANT, ), ) refused = await db.scalar( select(func.count(AuditLog.id)).where( AuditLog.action == ACTION_TOOL_REFUSED, AuditLog.created_at >= since, ), ) confirmed = await db.scalar( select(func.count(AuditLog.id)).where( AuditLog.action == ACTION_PROPOSAL_CONFIRMED, AuditLog.created_at >= since, ), ) # Card taps that were turned away — a forwarded card tapped by a colleague, a # leaver's still-live card, a tap after the release gate closed. Every one is # audited with ``status="denied"``, and until now NOTHING read those rows: the # impersonation the write path exists to block was invisible on the very page # whose own docstring calls a refusal a first-class number. Blocked and # unrecorded is only half a fix; blocked and unreadable is the other half. from app.assistant.card_actions import ( # noqa: PLC0415 — cycle ACTION_CARD_REFUSED, ) refused_card_taps = await db.scalar( select(func.count(AuditLog.id)).where( AuditLog.action == ACTION_CARD_REFUSED, AuditLog.created_at >= since, ), ) # ``tool_calls`` is a JSON array, so the counting cannot be pushed into SQL — # but the scan can be BOUNDED. This is a stats endpoint the admin page calls # on open, ``days`` goes up to 365, and the whole result is a top-10 list; an # unbounded fetch of every tool-calling message in a year to build it is the # same shape as the 200-round-trip list this module already fixed one screen # over. Newest first, because a top-10 of recent tools is what the number is # for, and the cap is logged rather than silently applied. tool_counts: dict[str, int] = {} rows = ( await db.execute( select(AssistantMessage.tool_calls) .where( AssistantMessage.created_at >= since, AssistantMessage.tool_calls.is_not(None), ) .order_by(AssistantMessage.created_at.desc()) .limit(_TOOL_SCAN_LIMIT), ) ).all() if len(rows) == _TOOL_SCAN_LIMIT: logger.info( "assistant: tool-usage counted over the most recent %d messages " "(window has at least that many)", _TOOL_SCAN_LIMIT, ) for (calls,) in rows: for name in calls or []: tool_counts[str(name)] = tool_counts.get(str(name), 0) + 1 # Deflection counts tickets NOT filed, which cannot tell a satisfied # employee from one who gave up. This is the only number that can. verdict_rows = ( await db.execute( select( AssistantMessageFeedback.verdict, func.count(AssistantMessageFeedback.id), ) .where(AssistantMessageFeedback.created_at >= since) .group_by(AssistantMessageFeedback.verdict), ) ).all() verdicts = {str(k): int(v) for k, v in verdict_rows} helpful = verdicts.get(VERDICT_HELPFUL, 0) unhelpful = verdicts.get(VERDICT_UNHELPFUL, 0) rated = helpful + unhelpful return { "window_days": days, "conversations": int(conversations or 0), "people": int(people or 0), "outcomes": {str(k or "unknown"): int(v) for k, v in outcome_rows}, "avg_latency_ms": int(latency) if latency is not None else None, "refused_tool_calls": int(refused or 0), "refused_card_taps": int(refused_card_taps or 0), "confirmed_proposals": int(confirmed or 0), "top_tools": [ {"name": name, "calls": calls} for name, calls in sorted( tool_counts.items(), key=lambda pair: (-pair[1], pair[0]), )[:10] ], "feedback": { "helpful": helpful, "unhelpful": unhelpful, "rated": rated, # None rather than 0 when nobody rated: a 0% helpful rate and "no # data yet" are different findings, and a dashboard that shows the # first for the second invents a problem. "helpful_rate": round(helpful / rated, 3) if rated else None, }, } async def submission_funnel(db: AsyncSession, *, days: int = 30) -> dict[str, Any]: """The funnel, end to end — and two things it refuses to count. **"No ticket was created" is not "auto-resolved."** The old headline number counted a conversation that produced nothing as a success. Here, a request is only ``closed_verified`` when a real business object reached ``effective``; ``closed_on_word`` (the employee said it helped and there was nothing to verify) is reported beside it, never merged into it. **"No feedback" is not success.** ``feedback_coverage`` is reported so the helpfulness rate can be read against how much of the population it describes; both are ``None`` rather than ``0`` when nobody rated, because a 0% rate and no data are different findings. Every number here is derived at read time from the links and the requests — nothing is a stored counter that could drift from what actually happened. """ from app.assistant.models import ( AssistantActionProposal, AssistantObjectLink, ) from app.core.coordinators.assistant_outcome_coordinator import ( PHASE_EFFECTIVE, PHASE_FAILED, PHASE_REJECTED, ) from app.core.coordinators.assistant_support_coordinator import ( OUTCOME_ACTION_NEEDS_ATTENTION, OUTCOME_ANSWER_HELPFUL_ACTION_PENDING, OUTCOME_SELF_SOLVED, OUTCOME_SYSTEM_VERIFIED, ) from app.domains.support.models import ( ORIGIN_ASSISTANT, REQUEST_ACTIVE, SupportServiceRequest, ) days = max(1, min(days, 365)) since = datetime.now(UTC) - timedelta(days=days) # ── proposals: offered → pressed → succeeded ────────────────────── # # ``attempt_count`` is what makes the middle number real. Before it, a # proposal that was pressed and failed was byte-identical to one nobody # touched, so "offered but never used" absorbed both. proposal_rows = ( await db.execute( select( AssistantActionProposal.status, func.count(AssistantActionProposal.id), func.sum( func.if_(AssistantActionProposal.attempt_count > 0, 1, 0) ), func.sum( func.if_( AssistantActionProposal.last_error_code.isnot(None), 1, 0 ) ), ) .where(AssistantActionProposal.created_at >= since) .group_by(AssistantActionProposal.status) ) ).all() offered = sum(int(r[1] or 0) for r in proposal_rows) pressed = sum(int(r[2] or 0) for r in proposal_rows) press_failed = sum(int(r[3] or 0) for r in proposal_rows) executed = sum( int(r[1] or 0) for r in proposal_rows if r[0] == "executed" ) # ── links: submitted → what became of them ──────────────────────── phase_rows = ( await db.execute( select(AssistantObjectLink.phase, func.count(AssistantObjectLink.id)) .where(AssistantObjectLink.created_at >= since) .group_by(AssistantObjectLink.phase) ) ).all() by_phase = {str(k or "unresolved"): int(v) for k, v in phase_rows} submitted = sum(by_phase.values()) miss_rows = ( await db.execute( select( AssistantActionProposal.link_miss_reason, func.count(AssistantActionProposal.id), ) .where( AssistantActionProposal.created_at >= since, AssistantActionProposal.link_miss_reason.isnot(None), ) .group_by(AssistantActionProposal.link_miss_reason) ) ).all() # ── service requests: how they ended, and on whose word ─────────── request_rows = ( await db.execute( select( SupportServiceRequest.status, SupportServiceRequest.outcome, func.count(SupportServiceRequest.id), ) .where( SupportServiceRequest.origin == ORIGIN_ASSISTANT, SupportServiceRequest.created_at >= since, ) .group_by(SupportServiceRequest.status, SupportServiceRequest.outcome) ) ).all() requests_total = sum(int(r[2] or 0) for r in request_rows) by_outcome: dict[str, int] = {} for _status, outcome, count in request_rows: by_outcome[str(outcome or "none")] = ( by_outcome.get(str(outcome or "none"), 0) + int(count or 0) ) needs_attention = sum( int(r[2] or 0) for r in request_rows if r[0] == REQUEST_ACTIVE and r[1] == OUTCOME_ACTION_NEEDS_ATTENTION ) # ── 从需求到最终结果的时间 ────────────────────────────────────── # # No new column: the need starts when the service request was opened and # ends when its link settled, and both timestamps already exist. Derived # rather than stamped so it can be recomputed over history — a counter # added today would have answered "unknown" for everything before today. # # Only links that actually SETTLED are measured. Averaging in the ones # still in flight would report a number that improves every time something # gets stuck, which is the wrong direction for the thing being measured. settle_rows = ( await db.execute( select( func.timestampdiff( text("HOUR"), SupportServiceRequest.created_at, AssistantObjectLink.phase_at, ) ) .select_from(AssistantObjectLink) .join( SupportServiceRequest, SupportServiceRequest.id == AssistantObjectLink.service_request_id, ) .where( AssistantObjectLink.created_at >= since, AssistantObjectLink.phase_at.isnot(None), AssistantObjectLink.phase.in_( (PHASE_EFFECTIVE, PHASE_REJECTED, PHASE_FAILED) ), ) ) ).scalars().all() settled_hours = sorted(int(h) for h in settle_rows if h is not None) def _median(values: list[int]) -> float | None: if not values: return None mid = len(values) // 2 if len(values) % 2: return float(values[mid]) return (values[mid - 1] + values[mid]) / 2 # ── card replies, grounding, escalation, feedback coverage ───────── # # All four come off the audit spine rather than a counter, for the same # reason: a counter added today cannot describe last month. audit_rows = ( await db.execute( select(AuditLog.action, func.count(AuditLog.id)) .where( AuditLog.created_at >= since, AuditLog.action.in_( (ACTION_TURN, "assistant.escalated", "assistant.card.reply"), ), ) .group_by(AuditLog.action) ) ).all() audit_counts = {str(a): int(n or 0) for a, n in audit_rows} turns = audit_counts.get(ACTION_TURN, 0) # ``is_card_reply`` / ``card_context_resolved`` were added to the turn audit # so that "nobody replies to cards" and "people reply but we cannot resolve # which card" stay distinguishable — they look identical from outside and # have opposite fixes. card_rows = ( await db.execute( select( func.sum( func.if_( func.json_unquote( func.json_extract(AuditLog.detail, "$.is_card_reply") ) == "true", 1, 0, ) ), func.sum( func.if_( func.json_unquote( func.json_extract( AuditLog.detail, "$.card_context_resolved" ) ) == "true", 1, 0, ) ), ).where(AuditLog.created_at >= since, AuditLog.action == ACTION_TURN) ) ).first() card_replies = int((card_rows[0] if card_rows else 0) or 0) card_resolved = int((card_rows[1] if card_rows else 0) or 0) # ── knowledge-gap reopen rate ────────────────────────────────────── reopen_rows = ( await db.execute( select(AuditLog.action, func.count(AuditLog.id)) .where( AuditLog.created_at >= since, AuditLog.action.in_( ( "assistant.knowledge_gap.resolve", "assistant.knowledge_gap.reopen", ), ), ) .group_by(AuditLog.action) ) ).all() gap_counts = {str(a): int(n or 0) for a, n in reopen_rows} gaps_resolved = gap_counts.get("assistant.knowledge_gap.resolve", 0) gaps_reopened = gap_counts.get("assistant.knowledge_gap.reopen", 0) def _rate(num: int, den: int) -> float | None: # ``None``, never 0, when there is nothing to divide by: a 0% rate and # "no data yet" are different findings, and showing the first for the # second invents a problem that does not exist. return round(num / den, 3) if den else None return { "window_days": days, "timing": { # Median and p90 rather than a mean: one request that sat for three # weeks would drag an average somewhere no individual request was. "settled_count": len(settled_hours), "median_hours_to_result": _median(settled_hours), "p90_hours_to_result": ( float(settled_hours[min(len(settled_hours) - 1, int(len(settled_hours) * 0.9))]) if settled_hours else None ), }, "engagement": { "turns": turns, "card_replies": card_replies, # Of the replies that came in ON a card, how many could we tell # WHICH card. A low number here is a plumbing problem; a low # ``card_replies`` is a discoverability problem. "card_replies_resolved": card_resolved, "card_reply_resolution_rate": _rate(card_resolved, card_replies), "escalations": audit_counts.get("assistant.escalated", 0), "escalation_rate": _rate(audit_counts.get("assistant.escalated", 0), turns), }, "knowledge_gaps": { "resolved": gaps_resolved, "reopened": gaps_reopened, # A closure that a real employee later contradicted. High is not # automatically bad — it means people are re-reporting rather than # giving up — but it must be visible either way. "reopen_rate": _rate(gaps_reopened, gaps_resolved), }, "proposals": { "offered": offered, # Pressed at least once — the number that did not exist before. "pressed": pressed, # Pressed, and the LAST attempt failed. A successful retry clears # ``last_error_code``, so this is "currently stuck", not "ever # failed" — which is the number worth acting on, and is why it is # not summed with ``executed`` to reach ``pressed``. "press_failed": press_failed, "executed": executed, }, "submissions": { "linked": submitted, "by_phase": by_phase, "effective": by_phase.get(PHASE_EFFECTIVE, 0), "rejected": by_phase.get(PHASE_REJECTED, 0), "failed": by_phase.get(PHASE_FAILED, 0), # A successful execution that linked nothing. Counted separately so # a gap in the outcome contract cannot hide inside "nobody # submitted anything" — the two look identical otherwise. "link_misses": {str(k): int(v) for k, v in miss_rows}, }, "requests": { "total": requests_total, "by_outcome": by_outcome, # Verified against a real object, versus taken on the employee's # word with nothing to check. Never added together. "closed_verified": by_outcome.get(OUTCOME_SYSTEM_VERIFIED, 0), "closed_on_word": by_outcome.get(OUTCOME_SELF_SOLVED, 0), "action_pending": by_outcome.get( OUTCOME_ANSWER_HELPFUL_ACTION_PENDING, 0 ), "needs_attention": needs_attention, }, } async def unhelpful_answers( db: AsyncSession, *, days: int = 30, limit: int = 50, status: str | None = None, ) -> list[dict[str, Any]]: """The answers employees said did not work, with what they had asked. The pairing is the point: 「没用」 on its own is a number, while the question beside it is a knowledge gap somebody can close. """ days = max(1, min(days, 365)) limit = max(1, min(limit, 200)) since = datetime.now(UTC) - timedelta(days=days) answers = aliased(AssistantMessage) stmt = select( AssistantMessageFeedback.created_at, AssistantMessageFeedback.note, AssistantMessageFeedback.reason_code, AssistantMessageFeedback.classification, AssistantMessageFeedback.gap_status, AssistantMessageFeedback.assigned_person_id, AssistantMessageFeedback.resolved_at, AssistantMessageFeedback.resolution_note, answers.id, answers.content, answers.tool_calls, answers.conversation_id, answers.seq, ).join(answers, answers.id == AssistantMessageFeedback.message_id).where( AssistantMessageFeedback.verdict == VERDICT_UNHELPFUL, AssistantMessageFeedback.created_at >= since, ) if status and status in (GAP_OPEN, GAP_IN_PROGRESS, GAP_RESOLVED, GAP_IGNORED): stmt = stmt.where(AssistantMessageFeedback.gap_status == status) rows = (await db.execute( stmt .order_by(AssistantMessageFeedback.created_at.desc()) .limit(limit) )).all() # The questions in one query, not one per row: ``limit`` reaches 200, and # 200 round trips to render a gap list is a page that feels broken. wanted = {(conv_id, max(int(seq) - 1, 0)) for *_, conv_id, seq in rows} questions: dict[tuple[Any, int], str] = {} if wanted: asked_rows = ( await db.execute( select( AssistantMessage.conversation_id, AssistantMessage.seq, AssistantMessage.content, ).where( tuple_( AssistantMessage.conversation_id, AssistantMessage.seq, ).in_(list(wanted)), ), ) ).all() questions = {(c, int(sq)): (body or "") for c, sq, body in asked_rows} out: list[dict[str, Any]] = [] for created_at, note, reason_code, classification, gap_status, assigned_id, resolved_at, resolution_note, msg_id, content, tools, conv_id, seq in rows: asked = questions.get((conv_id, max(int(seq) - 1, 0))) out.append({ "rated_at": created_at.isoformat() if created_at else None, "message_id": str(msg_id), "conversation_id": str(conv_id), "question": asked or "", "answer": content or "", "tools": list(tools or []), "note": note, "reason_code": reason_code, "classification": classification, "status": gap_status or GAP_OPEN, "assigned_person_id": str(assigned_id) if assigned_id else None, "resolved_at": resolved_at.isoformat() if resolved_at else None, "resolution_note": resolution_note, }) return out async def replay_gap( db: AsyncSession, *, message_id: uuid.UUID, actor_id: uuid.UUID, trace_id: str | None = None, ) -> dict[str, Any] | None: """Ask the gap's original question again, as the person who asked it. Records the shape of the turn and whether it passed. Does **not** close the gap: the person doing the work still has to say what they fixed, and a passing replay with no ``fix_type`` would let "it works now, somehow" count as a resolution. The first replay on a gap is also stored as ``verification_before`` — a replay run before the fix is the baseline, and without one "it cites a source now" cannot be distinguished from "it always did". """ from app.assistant.gap_verification import replay_question row = await db.scalar(select(AssistantMessageFeedback).where( AssistantMessageFeedback.message_id == message_id, AssistantMessageFeedback.verdict == VERDICT_UNHELPFUL, )) if row is None: return None # Read as columns, not through the ORM instance. A commit earlier in the # request expires every instance, and touching an expired attribute while # SQLAlchemy is compiling the next query fires the lazy load inside its own # cursor setup — which surfaces as MissingGreenlet rather than as anything # resembling the actual problem. located = ( await db.execute( select(AssistantMessage.conversation_id, AssistantMessage.seq) .where(AssistantMessage.id == message_id) ) ).first() if located is None or located[0] is None: return None conversation_id, answer_seq = located # The employee's own question, not the assistant's answer to it: the # question is what has to work now. Ordered by ``seq`` rather than # ``created_at`` — the transcript's own order, and two messages in one # exchange can share a timestamp. question = await db.scalar( select(AssistantMessage.content) .where( AssistantMessage.conversation_id == conversation_id, AssistantMessage.role == ROLE_USER, AssistantMessage.seq < answer_seq, ) .order_by(AssistantMessage.seq.desc()) .limit(1) ) if not question: return None asker_id = await db.scalar( select(AssistantConversation.person_id).where( AssistantConversation.id == conversation_id, ) ) if asker_id is None: return None shape = await replay_question( db, question=str(question), asker_id=asker_id, trace_id=str(trace_id or uuid.uuid4()), ) # "Passed" means the turn produced a grounded answer rather than a refusal # or an escalation. Deliberately mechanical: a human decides whether the # answer is *right*, and records that by choosing a fix type. What this # decides is only whether the assistant got anywhere at all — which is the # part that can be checked without reading somebody's data. # ``OUTCOME_ANSWERED`` is the only non-failure in the vocabulary — the # other two are ``unavailable`` and ``failed``. The extra condition is what # separates "answered" from "answered out of thin air": a turn that called # no tool and cited no source has not been grounded by the fix. passed = shape.outcome == OUTCOME_ANSWERED and bool( shape.tool_calls or shape.sources ) payload = shape.as_dict() if row.verification_before is None: row.verification_before = payload row.verification_after = payload row.verification_passed = passed row.verification_at = datetime.now(UTC).replace(tzinfo=None) await AuditService(db).log( trace_id=str(trace_id or "assistant.knowledge_gap.replay"), actor_type="user", action="assistant.knowledge_gap.replay", status="success", actor_user_id=actor_id, resource_type="assistant_message_feedback", resource_id=str(row.id), detail={"message_id": str(message_id), "passed": passed, **payload}, ) await db.commit() return {"passed": passed, "shape": payload} async def regression_set( db: AsyncSession, *, limit: int = 200, ) -> list[dict[str, Any]]: """Every question that failed, was fixed, and must keep passing. Kept as a set rather than a memory: the next change to retrieval or to a tool description can be checked against all of them, instead of against whichever one somebody happens to recall. """ rows = list( ( await db.execute( select(AssistantMessageFeedback, AssistantMessage) .join( AssistantMessage, AssistantMessage.id == AssistantMessageFeedback.message_id, ) .where(AssistantMessageFeedback.in_regression_set.is_(True)) .order_by(AssistantMessageFeedback.resolved_at.desc()) .limit(max(1, min(limit, 500))) ) ).all() ) out: list[dict[str, Any]] = [] for feedback, message in rows: out.append( { "message_id": str(feedback.message_id), "fix_type": feedback.fix_type, "evidence_ref": feedback.evidence_ref, "verification_passed": feedback.verification_passed, "verification_at": feedback.verification_at, "resolved_at": feedback.resolved_at, "conversation_id": ( str(message.conversation_id) if message.conversation_id else None ), } ) return out async def update_gap( db: AsyncSession, *, message_id: uuid.UUID, actor_id: uuid.UUID, action: str, note: str | None = None, trace_id: str | None = None, fix_type: str | None = None, evidence_ref: str | None = None, ) -> dict[str, Any] | None: """Apply one legal queue transition; repeated taps are idempotent. ``resolve`` now needs more than a sentence. See :mod:`app.assistant.gap_verification` for why each part is required and which fix types are exempt from which. ``ignore`` is unchanged and deliberately so: it means "this is not a gap" (a mistap, a duplicate, noise), which is a judgement about the *report*. Demanding a replay for it would only push people to close real gaps as ignored to get past the bar. """ row = await db.scalar(select(AssistantMessageFeedback).where( AssistantMessageFeedback.message_id == message_id, AssistantMessageFeedback.verdict == VERDICT_UNHELPFUL, )) if row is None: return None now = datetime.now(UTC).replace(tzinfo=None) if action == "claim": if row.gap_status in (GAP_RESOLVED, GAP_IGNORED): return {"status": row.gap_status, "assigned_person_id": str(row.assigned_person_id) if row.assigned_person_id else None} row.gap_status = GAP_IN_PROGRESS row.assigned_person_id = actor_id elif action in ("resolve", "ignore"): resolution_note = (note or "").strip() if not resolution_note: raise ValueError("A resolution note is required to close a knowledge gap") if action == "resolve": from app.assistant.gap_verification import ( FIX_TYPES_REQUIRING_REPLAY, assert_closure_is_supported, ) chosen_fix = (fix_type or row.fix_type or "").strip() or None chosen_evidence = ( (evidence_ref or "").strip() or row.evidence_ref or None ) # Raises with a code the router maps; nothing is written unless the # evidence supports the closure. assert_closure_is_supported( fix_type=chosen_fix, evidence_ref=chosen_evidence, verification_passed=row.verification_passed, ) row.fix_type = chosen_fix row.evidence_ref = chosen_evidence # A question that failed, was fixed, and now passes is worth keeping # for the next change to retrieval or a tool. ``out_of_scope`` is # not: there is nothing to re-verify, and a permanently-failing # entry would train everyone to ignore the set. if chosen_fix in FIX_TYPES_REQUIRING_REPLAY: row.in_regression_set = True row.gap_status = GAP_RESOLVED if action == "resolve" else GAP_IGNORED row.resolved_by_person_id = actor_id row.resolved_at = now row.resolution_note = resolution_note elif action == "reopen": row.gap_status = GAP_OPEN row.resolved_by_person_id = None row.resolved_at = None row.resolution_note = None # The fix, its evidence, and both turn shapes are NOT cleared: # reopening says the gap is back, not that the previous attempt never # happened, and the next person needs to see what was already tried. # # ``verification_passed`` IS cleared, and that is the whole point. A # real employee has just contradicted that replay. Leaving it set would # let the next closure pass on evidence that has already been shown to # be wrong — re-asking is exactly what has to happen again. row.verification_passed = None else: raise ValueError("Unknown knowledge-gap action") await AuditService(db).log( trace_id=trace_id or "assistant.knowledge_gap", actor_type="user", action=f"assistant.knowledge_gap.{action}", status="success", actor_user_id=actor_id, resource_type="assistant_message_feedback", resource_id=str(row.id), detail={ "message_id": str(message_id), "status": row.gap_status, "note": row.resolution_note, "fix_type": row.fix_type, "evidence_ref": row.evidence_ref, "verification_passed": row.verification_passed, }, ) await db.commit() return {"status": row.gap_status, "assigned_person_id": str(row.assigned_person_id) if row.assigned_person_id else None}