"""Closing a knowledge gap by proving it, not by describing it. Today a gap closes when somebody types a sentence into ``resolution_note``. That sentence is the entire evidence. It is written by the person who did the work, it is never checked, and the queue empties whether or not the assistant can now answer the question that opened the gap. Two things are required here instead: **A fix type, from a fixed list.** "Fixed it" is not a category. What was actually wrong is one of five things, and each has a different follow-up: a missing document, a retrieval or tool-description problem, a tool that returns the wrong field or holds bad data, a broken action or outcome link, or a request the product genuinely does not serve. A gap closed as :data:`FIX_OUT_OF_SCOPE` is a product decision; one closed as :data:`FIX_KB_DOC` is a document somebody has to keep current. Collapsing them loses the only information that makes the queue worth keeping. **A replay that passed.** The original question, asked again, as the original asker — so the same RBAC boundary applies and the answer is the one that employee would actually get, not the one an admin can see. A verification that fails cannot close the gap. The replay runs through the **dry-run simulation**, not a live turn — it must not be able to send a card or file a request. Re-asking a three-week-old question through the real path would push an unsolicited confirmation card at a colleague who never asked for it. What the replay records is deliberately *not* the answer text. The reply is that employee's own data rendered for them; storing it in an admin-visible queue would quietly turn a triage tool into a second copy of everybody's personal information. What is stored is the **shape** of the turn — which tools ran, which sources were cited, what the outcome was — which is what tells you whether the fix worked, and is the same before and after regardless of whose data flowed through it. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from typing import Any from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # ── what was actually wrong ────────────────────────────────────────── #: A document now exists (or was corrected) in the knowledge base. FIX_KB_DOC = "kb_doc" #: The content was there; the assistant could not find it, or the tool's own #: description did not describe what it was for. FIX_RETRIEVAL = "retrieval_or_description" #: A tool answered, but with the wrong field — or the data behind it was wrong. FIX_TOOL_DATA = "tool_field_or_data" #: The assistant could not act, or acted and could not report what happened. FIX_ACTION = "action_or_outcome_tracking" #: Confirmed: the platform does not do this, and is not going to. A decision, #: recorded as one, rather than a gap quietly marked resolved. FIX_OUT_OF_SCOPE = "out_of_scope" FIX_TYPES: tuple[str, ...] = ( FIX_KB_DOC, FIX_RETRIEVAL, FIX_TOOL_DATA, FIX_ACTION, FIX_OUT_OF_SCOPE, ) #: Fix types that must point at something. ``out_of_scope`` is the exception — #: the evidence for "we decided not to" is the decision itself, and demanding a #: URL for it would only teach people to paste a placeholder. FIX_TYPES_REQUIRING_EVIDENCE: frozenset[str] = frozenset( {FIX_KB_DOC, FIX_RETRIEVAL, FIX_TOOL_DATA, FIX_ACTION} ) #: Fix types whose claim is testable by asking the question again. #: ``out_of_scope`` is not: a replay would "fail" forever, and requiring one #: would make the honest answer the only one that cannot be recorded. FIX_TYPES_REQUIRING_REPLAY: frozenset[str] = frozenset( {FIX_KB_DOC, FIX_RETRIEVAL, FIX_TOOL_DATA, FIX_ACTION} ) class GapClosureError(ValueError): """A closure that does not meet the bar. Carries a code the API can map.""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code @dataclass(frozen=True) class TurnShape: """What a turn *did*, with none of what it said. The reply is the asker's own data. This is the part that answers "did the fix work" and is safe to keep in a queue other people read. """ outcome: str tool_calls: list[str] sources: list[str] offered_action: bool #: How many policy conflicts the turn surfaced. A replay that now flags one #: is a different finding from one that simply answered, and it is the #: signal a ``tool_field_or_data`` fix is most likely to move. policy_conflicts: int #: Length only. Enough to notice "it now says something" versus "it still #: says nothing", without keeping the something. reply_chars: int def as_dict(self) -> dict[str, Any]: return { "outcome": self.outcome, "tool_calls": self.tool_calls, "sources": self.sources, "offered_action": self.offered_action, "policy_conflicts": self.policy_conflicts, "reply_chars": self.reply_chars, } def shape_of(result: Any) -> TurnShape: """Reduce a :class:`SimulationResult` to what may be stored. ``docs`` is captured by the turn from the tool results themselves rather than from whatever the model chose to repeat, so a citation here means the source was really read. """ docs = getattr(result, "docs", None) or [] sources = [] for doc in docs: if isinstance(doc, dict): ref = doc.get("id") or doc.get("title") or doc.get("url") if ref: sources.append(str(ref)[:200]) reply = str(getattr(result, "reply", "") or "") tool_calls = [str(t)[:64] for t in (getattr(result, "tool_calls", None) or [])] if getattr(result, "turn_failed", False): outcome = "failed" elif not getattr(result, "available", True): outcome = "unavailable" else: outcome = "answered" return TurnShape( outcome=outcome, tool_calls=tool_calls, sources=sources, # A proposal is the difference between "explained it" and "did it", and # for a FIX_ACTION gap it is the whole claim being tested. The dry-run # dispatcher records what it *would* have offered without offering it. offered_action=bool(getattr(result, "proposed_actions", None)), policy_conflicts=len(getattr(result, "policy_conflicts", None) or []), reply_chars=len(reply), ) async def replay_question( db: AsyncSession, *, question: str, asker_id: uuid.UUID, trace_id: str, ) -> TurnShape: """Ask the original question again, as the person who originally asked it. **As them, not as the admin.** The assistant's tools are scoped to the caller; an admin replay would read a wider world and could pass on data the employee cannot see. The point is "can *they* get an answer now", and only their own boundary answers that. **Through the dry-run simulation, not the live turn.** ``simulate_employee_turn`` runs the production Feishu prompt, read bridge, action discovery and proposal validation, and substitutes exactly one thing: a dry-run proposal dispatcher. That matters twice over. It cannot send a card — otherwise re-asking a three-week-old question would push an unsolicited confirmation card at a colleague who never asked for it, and re-asking one about a laptop could file them a second request. And it writes no transcript, so replays cannot inflate the conversation counts, the funnel, or the admin transcript browser with turns no employee ever sent. ``trace_id`` is accepted for the audit record the caller writes; the simulation itself does not thread one. """ from app.assistant.conversation import simulate_employee_turn # noqa: PLC0415 from app.core.locale import resolve_lang # noqa: PLC0415 from app.identity.person_models import IdentityPerson # noqa: PLC0415 person = await db.get(IdentityPerson, asker_id) if person is None: raise GapClosureError( "ASSISTANT_GAP_ASKER_GONE", "The employee who reported this gap no longer exists, so the " "question cannot be replayed under their permissions.", ) # The asker's own language setting, so the replayed turn is the one they # would actually have received. result = await simulate_employee_turn( db, person=person, question=question, lang=resolve_lang(person), ) return shape_of(result) def assert_closure_is_supported( *, fix_type: str | None, evidence_ref: str | None, verification_passed: bool | None, ) -> None: """Refuse a closure the evidence does not support. Raises rather than returning a flag: every caller here must stop, and a boolean would eventually be checked in one place and not another. """ if not fix_type: raise GapClosureError( "ASSISTANT_GAP_FIX_TYPE_REQUIRED", "Choose what was actually wrong before closing this gap.", ) if fix_type not in FIX_TYPES: raise GapClosureError( "ASSISTANT_GAP_FIX_TYPE_INVALID", f"fix_type must be one of {list(FIX_TYPES)}.", ) if fix_type in FIX_TYPES_REQUIRING_EVIDENCE and not (evidence_ref or "").strip(): raise GapClosureError( "ASSISTANT_GAP_EVIDENCE_REQUIRED", "Link the fix — the published document, the change, or the tool it " "landed in.", ) if fix_type in FIX_TYPES_REQUIRING_REPLAY: if verification_passed is None: raise GapClosureError( "ASSISTANT_GAP_VERIFICATION_REQUIRED", "Re-ask the original question and record the result before " "closing this gap.", ) if not verification_passed: raise GapClosureError( "ASSISTANT_GAP_VERIFICATION_FAILED", "The replay did not pass. A gap whose question still fails is " "not resolved.", )